Custom Adapter

Pipelite exposes an open SPI that allows custom adapters to be created for any external system.

SPI dependency

<dependency>
    <groupId>io.pipelite</groupId>
    <artifactId>pipelite-spi</artifactId>
    <version>1.0.0-SNAPSHOT</version>
</dependency>

Implementing ChannelAdapter

public class MyCustomChannelAdapter implements ChannelAdapter {

    @Override
    public Endpoint createEndpoint(String url) {
        EndpointURL endpointURL = EndpointURL.parse(url);
        return new MyCustomEndpoint(endpointURL, this);
    }
}

Implementing Endpoint

public class MyCustomEndpoint extends DefaultEndpoint {

    public MyCustomEndpoint(EndpointURL url, ChannelAdapter adapter) {
        super(url, adapter);
    }

    @Override
    public Consumer createConsumer() {
        return new MyCustomConsumer(this);    // for use as source
    }

    @Override
    public Producer createProducer() {
        return new MyCustomProducer(this);    // for use as sink
    }
}

Implementing Consumer (source)

For event-driven consumers:

public class MyCustomConsumer extends AbstractConsumer
    implements EventDrivenConsumer {

    public void onMessage(Object data) {
        ExchangeImpl exchange = exchangeFactory.createExchange(data);
        process(exchange);
    }
}

For polling consumers:

public class MyCustomPollingConsumer extends DefaultPollingConsumer {

    @Override
    protected ExchangeImpl poll() {
        Object data = mySystem.fetchNext();
        if (data == null) return null;
        return exchangeFactory.createExchange(data);
    }
}

Implementing Producer (sink)

public class MyCustomProducer extends AbstractProducer {

    @Override
    public void produce(ExchangeImpl exchange) throws Exception {
        Object payload = exchange.getInputPayload();
        mySystem.send(payload);
    }
}

Registration via classpath scan

Create the registration file in the resources directory of your module:

src/main/resources/META-INF/pipelite/adapters

File content (one class per line):

com.example.adapters.MyCustomChannelAdapter

The FactoryComponentClasspathScanner automatically loads all classes registered in this file at PipeliteContext startup.

Integration with the context lifecycle

If your adapter needs to be notified of context startup and shutdown, implement ContextEventListener:

public class MyCustomChannelAdapter
    implements ChannelAdapter, ContextEventListener {

    @Override
    public void onContextStarted() {
        // start connection to the external system
    }

    @Override
    public void onContextStopped() {
        // graceful connection shutdown
    }
}