GH-3155: Add support for Java DSL extensions

[artembilan]

Fixes #3155

Provide an IntegrationFlowExtension for possible custom EI-operators
in the target project use-cases.

[artembilan]

@sirimamilla,

Please, take a look into this.
That's about your request for Java DSL extensions.

Pay attention to the test class changes:

public static class CustomIntegrationFlowDefinition
			extends IntegrationFlowExtension<CustomIntegrationFlowDefinition> {

		public CustomIntegrationFlowDefinition customEip() {
			return split()
					.transform("payload.toUpperCase()");
		}

	}

This is how your write your extension with any possible custom operators.
And this is how you need to use it:

@Bean
public IntegrationFlow customFlowDefinition() {
	return
		new CustomIntegrationFlowDefinition()
			.customEip()
			.aggregate()
			.get();
}

I'm not sure that we can come up with something much simpler.

I intentionally made it as a separate IntegrationFlowExtension<B extends IntegrationFlowExtension<B>> extends IntegrationFlowDefinition<B> class to make the statement as clear as possible: if you want an extension, please, implement your specific class. And this one is an extension of existing IntegrationFlowDefinition so you get access to all its methods for build any possible custom EI-operators.

Let me know WDYT?

/CC @garyrussell

[sirimamilla]

Hi @artembilan,
I will try this approach once and I will get back to you.

[garyrussell]

LGTM; nice!

[artembilan]

Thanks, Gary!

Let's wait for feedback from @sirimamilla anyway!

I see that some methods has to be moved to protected instead of private.
Plus might be the case that we need to make all the IntegrationComponentSpec implementation ctors as protected as well.
Let's imagine that we have missed some option to expose on the spec or we have a bug which is easy to overcome with an extension and override.
So, this way a custom operator against custom spect is going to be a good combination 😄

[garyrussell]

Let's wait ...

Agreed; I was just saying the approach looks good.

[sirimamilla]

Hi @artembilan,

This working as expected for the composite EIP patterns where in we need to extend the existing patterns. This is what I wanted to be able to do.

Let's imagine that we have missed some option to expose on the spec or we have a bug which is easy to overcome with an extension and override.

I agree, This would be good enhancement, as we may think some option is required for application but that may not fit good for framework itself. those features can be added using this

[sirimamilla]

Overall I am amazed by the way @artembilan solved this complex problem with very simple and effective solution.

[sirimamilla]

I was able to use the composite eip pattern and able to solve the requirement of selector on Gateway.

public <P> CustomIntegrationFlowDefinition selectingGateway(String channelName, Class<P> type, GenericSelector<P> selector) {
			MessageChannel aggregateChannel = new DirectChannel();
			return filter(type, selector, e -> e.discardChannel(aggregateChannel))
					.gateway(channelName)
					.channel(aggregateChannel);
		}

[artembilan]

Thanks, @sirimamilla , for feedback!

So, stay tuned and I'll finish this PR today!

I'm not sure that it will address all the possible extension peculiarities, but we can address them eventually (if needed)

[artembilan]

All docs fixed and rebased to master.

Thanks, Gary!

[sirimamilla]

HI @artembilan,

GatewayEndpointSpec(String requestChannel) can be made protected as well

GatewayEndpointSpec(String requestChannel) {
		super(new GatewayMessageHandler());
		this.handler.setRequestChannelName(requestChannel);
	}

[artembilan]

GatewayEndpointSpec(String requestChannel) can be made protected as well

Good catch! Thank you! Anything else ? 😉

[sirimamilla]

Good catch! Thank you! Anything else ? 😉

super(new GatewayMessageHandler()); please expose a constructor with this argument as protected. So extension classes will be able to extend the handler.

[artembilan]

So extension classes will be able to extend the handler.

I don't think so. That GatewayMessageHandler has a restricted responsibility to handle mid-flow requests. Nothing more.
I definitely not sure to what you can extend it.
But with the custom GatewayMessageHandler you are feel free to implement a custom ConsumerEndpointSpec on the matter similar to that GatewayEndpointSpec.

We talked about options to increase - the new component requires new contract and, therefore, new Spec. However with a new MessageHandler you might even don't need new spec - the .handle() fully covers you.

Please, elaborate more what is your goal to be able to extend that GatewayMessageHandler and therefore GatewayEndpointSpec?

[sirimamilla]

Hi @artembilan,

Here is an example of the use of allowing GatewayMessageHandler be extended. This will act as Selecting Gateway. It will preserve all the gateway behaviors along with the additional filter capability.

public static class CustomIntegrationFlowDefinition
			extends IntegrationFlowExtension<CustomIntegrationFlowDefinition> {

		public CustomIntegrationFlowDefinition upperCaseAfterSplit() {
			return split()
					.transform("payload.toUpperCase()");
		}

		public CustomIntegrationFlowDefinition customAggregate(Consumer<CustomAggregatorSpec> aggregator) {
			return register(new CustomAggregatorSpec(), aggregator);
		}

		public CustomIntegrationFlowDefinition selectingGateway(MessageChannel channel,
														Consumer<CustomGatewayEndpointSpec> endpointConfigurer) {
			return register(new CustomGatewayEndpointSpec(channel), endpointConfigurer);

		}

	}

	public static class CustomGatewayEndpointSpec extends GatewayEndpointSpec  {

		protected CustomGatewayEndpointSpec(MessageChannel requestChannel) {
			super(requestChannel);
			this.handler = new CustomGatewayMessageHandler();
			handler.setRequestChannel(requestChannel);

		}

		public CustomGatewayEndpointSpec selector(Predicate<Message<?>> condition) {
			if (this.handler instanceof CustomGatewayMessageHandler) {
				((CustomGatewayMessageHandler) this.handler).setCondition(condition);
			}
			return this;
		}
	}


	public static class CustomGatewayMessageHandler extends GatewayMessageHandler {

		Predicate<Message<?>> condition;

		public void setCondition(Predicate<Message<?>> condition) {
			this.condition = condition;
		}

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

			return condition.test(requestMessage) ? super.handleRequestMessage(requestMessage) : requestMessage;
		}
	}

Selecting Gateway can be used as below.

@Bean
		public IntegrationFlow selectingGateway() {
			return new CustomIntegrationFlowDefinition()
					.selectingGateway(customFlowDefinition().getInputChannel(),
							g -> g.selector(m -> m.getHeaders().containsKey("pass")))
					.get();
		}

This will help in doing any further enhancements to the GatewayMessageHandler.

[sirimamilla]

Hi @artembilan,

Protected actor can be added to below.

JmsPollableMessageChannelSpec(JmsChannelFactoryBean jmsChannelFactoryBean, ConnectionFactory connectionFactory) {
		this.jmsChannelFactoryBean = jmsChannelFactoryBean;
		this.jmsChannelFactoryBean.setConnectionFactory(connectionFactory);
		this.jmsChannelFactoryBean.setSingleton(false);
		this.jmsChannelFactoryBean.setBeanFactory(new DefaultListableBeanFactory());
	}

[artembilan]

This will help in doing any further enhancements to the GatewayMessageHandler.

That's abusing a purpose of all the components we have out-of-the-box. It is not preserving, it is some hybrid I'd like to avoid.
You feel free to do such a "hell" in your own project, but I'm not going to encourage all users of our project to follow your idea on the matter.
Again: such a "selective gateway" can be implemented with existing components in some separate flow which could be referenced from other flows.

Sorry not accepting your request, but I'm fully against some hybrids when go beyond EIP catalog.

The JmsPollableMessageChannelSpec feedback is accepted 😄

[sirimamilla]

That's abusing a purpose of all the components we have out-of-the-box. It is not preserving, it is some hybrid I'd like to avoid.

Even I agree with you, from EIP catalog this is an extension.

I will implement this in my application.

For this PR, I think we are almost ready to merge. No further comments from me.

[sirimamilla]

Hi @artembilan ,

As reviewed, PR is performing the tasks as expected and can be merged. :)

[artembilan]

and can be merged

So, @garyrussell , the final decision is up to you 😉