Spring Integration - DSL - Split или Fork - PullRequest
Новая
       22

Spring Integration - DSL - Split или Fork

0 голосов
/ 01 сентября 2018

Я хочу создать поток IntegrationFlow для потока ниже.

enter image description here

  1. От начала к доставке - поток синхронизации.
  2. Как извлечь / разветвить асинхронный конечный узел из элементов сборки и проверить элементы. 

	@Bean
	public IntegrationFlow buildCart() {
		return f -> f.handle(validate, "buildPreCheck")
.handle(preProcessProcessor)
.handle(getOffersProcessor)
.handle(buildItems)
**.wireTap(log())**
.handle(validateItems)
.handle(deliver);
	}

EDIT:

Привет, Артем, я добавил Wire Tap, как показано ниже. Тем не менее он исключает узел WireTap как Sequencal и ожидает этого узла.

Пожалуйста, помогите сделать его узлом Aysnc. 

@Bean
public IntegrationFlow log() {
    return f -> f.handle(auditProcessor).channel("nullChannel");
}



@ServiceActivator
@Description("Call and get the Offers Request")
public void getDetails(Message<Context> message) throws InterruptedException {
    log.info("getDetails-Sleep-Start");
    Thread.sleep(3000);
    log.info("getDetails-Sleep-End");
}

1 Ответ

0 голосов
/ 01 сентября 2018

В Java DSL Spring Integration нам всем не хватает того, что одним из очень важных компонентов в Spring Integration является MessageChannel. И дело в том, что каналы могут добавляться в поток всякий раз, когда нам нужно больше, чем значение по умолчанию DirectChannel. Для асинхронного выполнения у нас есть ExecutorChannel. Но прежде чем мы пойдем по этому разветвленному потоку, нам нужно как-то прыгнуть туда, не нарушая основного. С точки зрения EIP это называется Wire-Tap: https://www.enterpriseintegrationpatterns.com/patterns/messaging/WireTap.html.

Spring Integration Java DSL предлагает реализацию, подобную оператору .wireTap() в потоке. Логика аудита может быть реализована в подключенном потоке или по каналу, но не забывайте о ExecutorChannel!

Вы можете увидеть больше информации в Справочном руководстве: https://docs.spring.io/spring-integration/reference/html/java-dsl.html#java-dsl-wiretap

UPDATE 

.handle(buildItems)
.wireTap(log())

- правильный путь: вы собираетесь проверить результат buildItems и перейти к следующему шагу.

log() необходимо изменить следующим образом: 

@Bean
public IntegrationFlow log() {
    return f -> f.channel(c -> c.executor(taskExecutorBean())).handle(auditProcessor).channel("nullChannel");
}

Обратите внимание на c.executor(). Таким образом, мы добавляем асинхронную передачу для нашего подпотока log().

...