How to create custom component and add it to flow

Question

I want to create a customer component trace() and want to use it in the flow.

Something like CustomFlows.from("").trace().get();

Can you suggest me how to do it? Seems like IntegrationFlowDefinition is closed and not extendable.

Answer 1

That is an interesting technique... But right now I'm not convinced that we should allow such a extension for the FlowBuilder... Feel free to raise GitHub issue on the matter and we will see in the future how to be with that.

For such an extension you should understand how the Framework works on the background.

Let's consider the simplest component .bridge()! Its goal to transfer message from one channel to another. So, this component is of request-reply type and must be tied with those two channels on it sides:

.channel()
.bridge()
.channel()

Of course DSL can create DirectChannels on those places where we don't define explicit one. Or we can rely on the replyChannel header, too if we finish our flow with its end - .get(). And so on.

The BridgeHandler is an out-of-the-box component, that's why we provide the explicit DSL method for it.

But from other side it has very simple code:

public class BridgeHandler extends AbstractReplyProducingMessageHandler {

    @Override
    public String getComponentType() {
        return "bridge";
    }

    @Override
    protected Object handleRequestMessage(Message<?> requestMessage) {
        return requestMessage;
    }

    @Override
    protected boolean shouldCopyRequestHeaders() {
        return false;
    }
}

As you see it is a MessageHandler implementation, so we can use it directly like this:

.handle(new BridgeHandler())

From here the question from me to you: what is the reason to bring some complexity to the Framework, if you can reach your requirements with existing features: implement MessageHandler, extend AbstractPayloadTransformer. Or even just forget about them all and do everything with POJO method invocation, like this since DSL 1.1:

@Override
protected IntegrationFlowDefinition<?> buildFlow() {
        return from(this, "messageSource", e -> e.poller(p -> p.trigger(this::nextExecutionTime)))
                .split(this)
                .transform(this)
                .aggregate(a -> a.processor(this, null), null)
                .enrichHeaders(Collections.singletonMap("foo", "FOO"))
                .filter(this)
                .handle(this)
                .channel(c -> c.queue("myFlowAdapterOutput"));
    }

See my article on the matter.

So, you should find many argument to convince me to extend framework that way :-).