Skip to main content

Connector

A Connector represents a protocol capability: FIX, Kafka, an internal binary feed, a proprietary REST API. It is the bridge between the platform's pipeline and the outside world. A Connector both produces messages (handing inbound traffic into the pipeline) and consumes messages (accepting outbound traffic for delivery).

Reach for a Connector only when an official plugin does not already cover the protocol. Connectors are the most involved component type because they own a long-lived session, a thread, and (often) network state.

PropertyValue
Java interfacecom.connamara.sdk.v1.adapter.components.Connector
Methodsstart(Map<String, Object> configuration, AdapterContext context), stop(), isRunning(), plus the MessageProducer and MessageConsumer methods inherited from those interfaces.
Pipeline slotSource (produces inbound messages) or Target (accepts outbound messages). A single Connector class can play either role depending on how it's wired.
LifecycleThe platform calls start(...) once per configured connection at adapter boot, and stop() once at shutdown or reconfiguration. The implementation owns its own threads.
Side effectsRequired. Connectors hold sockets, sessions, and resources.

Skeleton​

A complete Connector is too long to inline. The contract below shows the four hooks every implementation provides; refer to the official plugins (FIX, Kafka, Essentials) under the plugins/ source tree for production-grade reference implementations.

package com.example.myplugin;

import com.connamara.sdk.v1.adapter.AdapterContext;
import com.connamara.sdk.v1.adapter.components.Connector;
import com.connamara.sdk.v1.common.component.ManifestResource;
import com.connamara.sdk.v1.common.message.Message;
import com.connamara.sdk.v1.common.message.MessageListener;

import java.util.Map;

@ManifestResource("""
    {
      "id": "my-protocol-connector",
      "pluginId": "my-plugin",
      "functionalType": "my-protocol-connector",
      "displayName": "My Protocol",
      "category": "CONNECTOR",
      "configuration": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "type": "object",
        "title": "My Protocol Settings",
        "x-order": ["host", "port"],
        "required": ["host", "port"],
        "properties": {
          "host": { "type": "string", "title": "Host" },
          "port": { "type": "integer", "title": "Port", "x-is-port": true }
        }
      }
    }
    """)
public class MyProtocolConnector implements Connector {

    private final Map<String, Object> configuration;
    private volatile boolean running = false;
    private MessageListener<Object> listener;
    // Hold session, socket, threads, etc. as instance state.

    public MyProtocolConnector(Map<String, Object> configuration) {
        this.configuration = configuration;
    }

    @Override
    public void start(Map<String, Object> connectionConfig, AdapterContext context) {
        // Open the session, start consumer threads, register inbound message dispatch.
        // On inbound traffic, hand the message to the platform via:
        //     listener.onMessage(message);
        running = true;
    }

    @Override
    public void stop() {
        // Close the session, drain queues, join threads.
        running = false;
    }

    @Override
    public boolean isRunning() {
        return running;
    }

    @Override
    public void send(Message<Object> message) {
        // Outbound: serialise the message onto the wire.
        // Called when the Connector is wired as a route's target.
    }

    @Override
    public void setListener(MessageListener<Object> listener) {
        // Inbound: the platform calls this once before start(...). Hold the
        // reference and call listener.onMessage(...) when traffic arrives.
        this.listener = listener;
    }

    // addObserver / removeObserver / getHealth / dispatch are inherited from
    // the parent interfaces; see the SDK for the full contract.
}

Key rules​

  • Threading is yours. A Connector owns its own threads. Use a single, named thread per session for long-lived I/O loops; do not block the constructor or start(...) waiting for the session to fully initialise. Return promptly and bring the session up asynchronously.
  • Inbound flow goes through the listener. When inbound traffic arrives, hand it to the platform by calling listener.onMessage(message) on the MessageListener registered via setListener(...). The platform routes the message into the configured pipeline from there.
  • Outbound flow comes through send(...). When the Connector is wired as a target, the platform calls send(...). Implementations must serialise the message and hand it off to the wire; the call returns when the message has been accepted by the underlying transport (not necessarily delivered to the counterparty).
  • Manage observers safely. Observers (additional MessageListener instances registered for diagnostic copies) can be added or removed concurrently with inbound dispatch. Use a copy-on-write list or take a snapshot before iterating, and never iterate the live observer collection while invoking onMessage.
  • Mark network ports in the manifest. Add "x-is-port": true to any port field in the manifest. The platform uses this to extract reservable ports for capacity planning.

When to reach for a Connector​

NeedReach for
Add a protocol the official plugins do not provide.Connector.
Add a transformation in front of an existing protocol.Transformer; reuse the existing Connector.
Add per-message logic in front of an existing protocol.Processor; reuse the existing Connector.
Mock or simulate an external system in tests.Reuse the test Connector pattern from the official plugin tests; building a production Connector for testing is rarely worth the effort.

See also​