Статьи

Выпущена весенняя интеграция Java DSL 1.0 RC1

Январь 16, 2018

Оригинальная статья написана Артемом Биланом. 


Уважаемое сообщество Spring,

Мы рады сообщить, что вскоре после  релиза-кандидата Spring Integration 4.1  доступен релиз-релиз Spring Integration Java DSL 1.0. Пожалуйста, используйте  Milestone Repository  с Maven или Gradle, или загрузите  дистрибутивный архив , чтобы получить его вращение.

Смотрите домашнюю страницу проекта   для получения дополнительной информации.

Релиз включает в себя множество новых функций и улучшений, а также ряд исправлений ошибок. Релиз GA запланирован на середину ноября.

Вот краткое изложение основных изменений с момента  последнего этапа :

Рефакторинг и критические изменения

Поддерживая более ранние версии Java, Java DSL Spring Integration в первую очередь ориентирован на Java 8 и его поддержку Lambda. Мы удалили несколько  functional interfaces в пользу аналогичных интерфейсов из Java 8:  Consumer<T>и  Function<T, R> т. Д. Конечно, чтобы поддержать обратную совместимость со старой версией Java, мы реализовали подобные интерфейсы в исходном коде DSL. Пользователи, которые используют измененные интерфейсы с версиями Java менее 8, должны будут внести изменения, чтобы исправить ошибки компиляции. Например:

Из этого:

.handle(Integer.class, (p, h) -> p * 2,
new EndpointConfigurer<GenericEndpointSpec<ServiceActivatingHandler>>() {
@Override
public void accept(GenericEndpointSpec<ServiceActivatingHandler> spec) {
spec.poller(Pollers.cron("7 * * * * ?"));
}
})

To this:

.handle(Integer.class, (p, h) -> p * 2,
new Consumer<GenericEndpointSpec<ServiceActivatingHandler>>() {
@Override
public void accept(GenericEndpointSpec<ServiceActivatingHandler> spec) {
spec.poller(Pollers.cron("7 * * * * ?"));
}
})

Of course if you use a Java 8 Lambda here, the code will not require changes:

.handle(Integer.class, (p, h) -> p * 2, e -> e.poller(Pollers.cron("7 * * * * ?")))

The IntegrationFlows now contains only from(...) methods. the.fromFixedMessageChannel() has been replaced with .from(String messageChannelName, boolean fixedSubscriber).

In addition, to fix some package tangle issues, we have moved some classes to different packages.

Method Scope Functions

To simplify the code completion from an IDE and allow avoiding redundant searches for a desired Namespace Factory we added overloaded methods with Function<T, R>argument. For example these code snippets are equal:

.....
.channel(Amqp.pollableChannel(this.rabbitConnectionFactory)
.queueName("amqpReplyChannel")
.channelTransacted(true))
....
.channel(c -> c.amqpPollable(this.rabbitConnectionFactory)
.queueName("amqpReplyChannel")
.channelTransacted(true))
....

Where the c variable is the Channel‘s «method-aggregator» object, which delegates to the appropriate Namespace Factory. Other similar Lambda methods are:

  • IntegrationFlows.from(MessageSourcesFunction sources)
  • IntegrationFlows.from(MessageProducersFunction producers)
  • IntegrationFlows.from(MessagingGatewaysFunction gateways)
  • IntegrationFlowDefinition.handleWithAdapter(Function<Adapters, MessageHandlerSpec<?, H>> adapters)
  • EndpointSpec.poller(Function<PollerFactory, PollerSpec> pollers)

FunctionExpression

Spring Integration has amazing Spring Expression Language (SpEL) support. Since the Java DSL is pure (eh!) Java, it does not really make sense to specify some business logic in a long String for an expression property. Being inspired by Java 8 Lambda support, and pursuing the aim of minimal changes we have introduced the FunctionExpression — an implementation of the SpEL Expression interface — which accepts a Function<T, R> and delegates to it on the each getValue(). Now, many components in the DSL provide(Function<T, R> function) methods as an alternative to the similar SpEL method. Here is an example for the localFilename property for theFtpInboundFileSynchronizingMessageSource:

With SpEL:

@Bean
public IntegrationFlow ftpInboundFlow() {
return IntegrationFlows
.from(s -> s.ftp(this.ftpSessionFactory)
.remoteDirectory("ftpSource")
.localFilenameExpression("payload.toUpperCase() + '.a'")
.channel(c -> c.queue("ftpInboundResultChannel"))
.get();
}

With Lambda:

@Bean
public IntegrationFlow ftpInboundFlow() {
return IntegrationFlows
.from(s -> s.ftp(this.ftpSessionFactory)
.remoteDirectory("ftpSource")
.localFilename(f -> f.toUpperCase() + ".a")))
.channel(c -> c.queue("ftpInboundResultChannel"))
.get();
}

Other interesting uses of the FunctionExpression are the Enricher andHeaderEnricher:

.enrich(e -> e.requestChannel("enrichChannel")
.requestPayload(Message::getPayload)
.propertyFunction("date", m -> new Date()))

The FunctionExpression also supports runtime type conversion as is done in the standardSpelExpression.

SubFlows

We have introduced SubFlow support for some if...else and publish-subscribecomponents. The simplest example is .publishSubscribeChannel():

@Bean
public IntegrationFlow subscribersFlow() {
return flow -> flow
.publishSubscribeChannel(Executors.newCachedThreadPool(), s -> s
.subscribe(f -> f
.<Integer>handle((p, h) -> p / 2)
.channel(c -> c.queue("subscriber1Results")))
.subscribe(f -> f
.<Integer>handle((p, h) -> p * 2)
.channel(c -> c.queue("subscriber2Results"))))
.<Integer>handle((p, h) -> p * 3)
.channel(c -> c.queue("subscriber3Results"));
}

Of course the same result we can be achieved with separate IntegrationFlow @Beandefinitions, but we hope you’ll find the subflow style of logic composition useful.

Similar publish-subscribe subflow composition is provided by .routeToRecipients().

Another example is .discardFlow() instead of .discardChannel() on .filter().

.route() deserves special attention:

@Bean
public IntegrationFlow routeFlow() {
return f -> f
.<Integer, Boolean>route(p -> p % 2 == 0,
m -> m.channelMapping("true", "evenChannel")
.subFlowMapping("false", sf ->
sf.<Integer>handle((p, h) -> p * 3)))
.transform(Object::toString)
.channel(c -> c.queue("oddChannel"));
}

The .channelMapping() continues to work as in regular Router mapping, but the.subFlowMapping() tied that subflow with main flow. In other words, any router’s subflow returns to the main flow after .route().

Similar «return-to-main-flow» subflow is supported by .gateway():

@Bean
public IntegrationFlow gatewayFlow() {
return f -> 
f.gateway("gatewayRequest", g -> g.errorChannel("gatewayError").replyTimeout(10L))
.gateway(gf -> gf.transform("From Gateway SubFlow: "::concat));
}

However this Gateway SubFlow is just wired with main flow through the explicitDirectChannel and wrapped to the regular GatewayMessageHandler using that channel as a requestChannel option.

Of course, subflows can be nested with any depth, but we don’t recommend to do that because, in fact, even in the router case, adding complex subflows within a flow would quickly begin to be difficult for a human to parse.

Conclusion

We haven’t added more protocol specific adapters since the last milestone. Not all adapters will be supported directly by the DSL although the most commonly used ones have first class support. However, those that don’t have first class support can easily be wired in using .handle(). As we have discussed previously, we are looking for input to prioritize the implementations of the remaining adapters so, don’t be shy to share your thoughts and ideas!

You can obtain more information about these and existing classes from their source code and from Reference Manual.

We look forward to your comments and feedback (StackOverflow (spring-integrationtag), Spring JIRA, GitHub) as soon as possible and report issues you find before we GA towards over a couple weeks.

As always, we very much welcome contributions.

28
Share: