Joule
  • Welcome to Joule's Docs
  • Why Joule?
    • Joule capabilities
  • What is Joule?
    • Key features
    • The tech stack
  • Use case enablement
    • Use case building framework
  • Concepts
    • Core concepts
    • Low code development
    • Unified execution engine
    • Batch and stream processing
    • Continuous metrics
    • Key Joule data types
      • StreamEvent object
      • Contextual data
      • GeoNode
  • Tutorials
    • Getting started
    • Build your first use case
    • Stream sliding window quote analytics
    • Advanced tutorials
      • Custom missing value processor
      • Stateless Bollinger band analytics
      • IoT device control
  • FAQ
  • Glossary
  • Components
    • Pipelines
      • Use case anatomy
      • Data priming
        • Types of import
      • Processing unit
      • Group by
      • Emit computed events
      • Telemetry auditing
    • Processors
      • Common attributes
      • Filters
        • By type
        • By expression
        • Send on delta
        • Remove attributes
        • Drop all events
      • Enrichment
        • Key concepts
          • Anatomy of enrichment DSL
          • Banking example
        • Metrics
        • Dynamic contextual data
          • Caching architecture
        • Static contextual data
      • Transformation
        • Field Tokeniser
        • Obfuscation
          • Encryption
          • Masking
          • Bucketing
          • Redaction
      • Triggers
        • Change Data Capture
        • Business rules
      • Stream join
        • Inner stream joins
        • Outer stream joins
        • Join attributes & policy
      • Event tap
        • Anatomy of a Tap
        • SQL Queries
    • Analytics
      • Analytic tools
        • User defined analytics
          • Streaming analytics example
          • User defined analytics
          • User defined scripts
          • User defined functions
            • Average function library
        • Window analytics
          • Tumbling window
          • Sliding window
          • Aggregate functions
        • Analytic functions
          • Stateful
            • Exponential moving average
            • Rolling Sum
          • Stateless
            • Normalisation
              • Absolute max
              • Min max
              • Standardisation
              • Mean
              • Log
              • Z-Score
            • Scaling
              • Unit scale
              • Robust Scale
            • Statistics
              • Statistic summaries
              • Weighted moving average
              • Simple moving average
              • Count
            • General
              • Euclidean
        • Advanced analytics
          • Geospatial
            • Entity geo tracker
            • Geofence occupancy trigger
            • Geo search
            • IP address resolver
            • Reverse geocoding
            • Spatial Index
          • HyperLogLog
          • Distinct counter
      • ML inferencing
        • Feature engineering
          • Scripting
          • Scaling
          • Transform
        • Online predictive analytics
        • Model audit
        • Model management
      • Metrics engine
        • Create metrics
        • Apply metrics
        • Manage metrics
        • Priming metrics
    • Contextual data
      • Architecture
      • Configuration
      • MinIO S3
      • Apache Geode
    • Connectors
      • Sources
        • Kafka
          • Ingestion
        • RabbitMQ
          • Further RabbitMQ configurations
        • MQTT
          • Topic wildcards
          • Session management
          • Last Will and Testament
        • Rest endpoints
        • MinIO S3
        • File watcher
      • Sinks
        • Kafka
        • RabbitMQ
          • Further configurations
        • MQTT
          • Persistent messaging
          • Last Will and Testament
        • SQL databases
        • InfluxDB
        • MongoDB
        • Geode
        • WebSocket endpoint
        • MinIO S3
        • File transport
        • Slack
        • Email
      • Serialisers
        • Serialisation
          • Custom transform example
          • Formatters
        • Deserialisers
          • Custom parsing example
    • Observability
      • Enabling JMX for Joule
      • Meters
      • Metrics API
  • DEVELOPER GUIDES
    • Setting up developer environment
      • Environment setup
      • Build and deploy
      • Install Joule
        • Install Docker demo environment
        • Install with Docker
        • Install from source
        • Install Joule examples
    • Joulectl CLI
    • API Endpoints
      • Mangement API
        • Use case
        • Pipelines
        • Data connectors
        • Contextual data
      • Data access API
        • Query
        • Upload
        • WebSocket
      • SQL support
    • Builder SDK
      • Connector API
        • Sources
          • StreamEventParser API
        • Sinks
          • CustomTransformer API
      • Processor API
      • Analytics API
        • Create custom metrics
        • Define analytics
        • Windows API
        • SQL queries
      • Transformation API
        • Obfuscation API
        • FieldTokenizer API
      • File processing
      • Data types
        • StreamEvent
        • ReferenceDataObject
        • GeoNode
    • System configuration
      • System properties
  • Deployment strategies
    • Deployment Overview
    • Single Node
    • Cluster
    • GuardianDB
    • Packaging
      • Containers
      • Bare metal
  • Product updates
    • Public Roadmap
    • Release Notes
      • v1.2.0 Join Streams with stateful analytics
      • v1.1.0 Streaming analytics enhancements
      • v1.0.4 Predictive stream processing
      • v1.0.3 Contextual SQL based metrics
    • Change history
Powered by GitBook
On this page
  • Development steps
  • Explaining each step
  • Step 1: Create project using the template
  • Step 2: Define connector specification
  • Step 3: Implement custom transport
  • Step 4: Build, test and package
  • Step 5: Deploy

Was this helpful?

  1. DEVELOPER GUIDES
  2. Builder SDK

Connector API

Connectors are key to enable Joule to perform use cases. Given the importance of Connectors, an API has been provided to enable developers to build and extend the capabilities of the platform.

PreviousBuilder SDKNextSources

Last updated 6 months ago

Was this helpful?

Development steps

  1. Create project using the template

  2. Define connector specification

  3. Implement custom transport

  4. Build, test and package

  5. Deploy

Explaining each step

Step 1: Create project using the template

We have provided a project template project to quick start development. The project can be found . Clone the template project and copy relevant code and structure to your own project

git clone git@gitlab.com:joule-platform/fractalworks-project-templates.git

Joule uses Gradle to manage Java dependencies. To add dependencies for your connector, manage them in the build.gradle file inside your connector's directory.

Step 2: Define connector specification

For components to be defined using the Joule DSL a specification class is required. Joule provides a AbstractTransportSpecification class that is to be extended for your implementation. This class provides core attributes that support a number of features such as:

  • Batching

  • Formatting

  • Parsing Threads

Publisher Example

@JsonRootName(value = "templatePublisher")
public class TemplatePublisherSpecification extends AbstractTransportSpecification {

    private String someField;

    /**
     * Default and required
     */
    public TemplatePublisherSpecification() {
        super();
    }

    /**
     * Default and required
     */
    public TemplatePublisherSpecification(String name) {
        super(name);
    }
    
    @JsonProperty(value = "some field", required = true)
    public void setSomeField(String someField) {
        this.someField = someField;
    }

    @Override
    public void validate() throws InvalidSpecificationException {
        super.validate();
        // TODO: Add validation logic based upon required fields
        if (someField == null || someField.isEmpty()) {
            throw new InvalidSpecificationException("someField must be provided.");
        }
    }
    
    @JsonIgnore
    @Override
    public Class<? extends Transport> getComponentClass() {
        // TODO: Change to your transport class
        return TemplatePublisherTransport.class;
    }
}

There is an option to create a specification builder in the event of programmactically. See the TemplatePublisherSpecificationBuilder example class for a reference implementation.

For Joule to load and initialised the component the specifications need to be defined within the plugins.properties file under the META-INF/services directory

Example

# Change and add lines for your specification classes
com.fractalworks.streams.examples.transport.TemplatePublisherSpecification

Step 3: Implement custom transport

Now we can move on to building the transport class that will either consume or publish events.

Publisher Example

Publisher has two key methods to implement; initialise and publish.

// Some code
public class TemplatePublisherTransport extends AbstractPublisherTransport {

    private String someField;

    public TemplatePublisherTransport() {
        super();
        logger = LoggerFactory.getLogger(TemplatePublisherTransport.class);
    }

    public TemplatePublisherTransport(TemplatePublisherSpecification specification) {
        super(specification);
        logger = LoggerFactory.getLogger(TemplatePublisherTransport.class);
        someField = specification.getSomeField();
    }

    @Override
    public void initialize() {
        super.initialize();
        // TODO: Add custom transport initialisation code
    }

    @Override
    public void publish(Collection<StreamEvent> events) {
        // TODO: Add custom transport logic to transmit events
    }
}

Consumer Example

Consumer has two key methods to implement; initialise and start.

public class TemplateConsumerTransport extends AbstractConsumerTransport {

    public TemplateConsumerTransport() {
        super();
        logger = LoggerFactory.getLogger(TemplateConsumerTransport.class);
    }

    public TemplateConsumerTransport(TemplateConsumerSpecification specification) {
        super(specification);
        logger = LoggerFactory.getLogger(TemplateConsumerTransport.class);
    }

    @Override
    public void initialize() {
        super.initialize();
        // TODO: Add custom transport initialisation code
    }
    

    @Override
    public void start() {
        super.start();
        // TODO: Add custom transport consumer code
    }
}

Step 4: Build, test and package

The template project provides basic JUnit test to validate DSL. The project will execute these tests during the gradle build cycle and deploy to your local maven repository.

gradle build publishToMavenLocal

Step 5: Deploy

cp build/libs/<your-connector>.jar <location>/userlibs

Once your package has been successfully created you are ready to deploy to a Joule project. The resulting jar artefact needs to be placed in to the userlibs directory in your Joule projects directory. See provided examples for further directions.

here
documentation