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
}
}