Spring интеграция маршрутизации в подпоток в другом классе

Question

Я борюсь за пару дней, чтобы сделать эту работу.То, что я пытаюсь сделать, - это вызывать различные подпотоки (которые являются потоком интеграции) из основного потока на основе содержимого сообщения и после его завершения, чтобы вернуться к основному потоку.Это похоже на делегирование ответственности определенному классу, чтобы закончить что-то и вернуться к основному потоку.Эта ответственность может также потребовать некоторых шагов, поэтому она также будет реализована.Вот мой основной поток:

public IntegrationFlow processingFlow(
  MessageChannel eventIn,
  MessageChannel eventOut,
  ChangedEventsLoader changedEventsLoader,
  CalculatorRouter calculatorRouter) {

return IntegrationFlows.from(eventIn)
    .handle(changedEventsLoader)
    .route(
        CalculatorRouter::getSportId,
        CalculatorRouter::routeCalculation)
    .channel(eventOut)
    .get();

}

Вот реализация маршрутизатора:

@Service
@AllArgsConstructor
public class CalculatorRouter {
  private final MessageChannel eventOut;

  public RouterSpec<Integer, MethodInvokingRouter> routeCalculation(
      RouterSpec<Integer, MethodInvokingRouter> mapping) {
    return mapping
        .channelMapping(1, "subflowCalculationChannel")
        .defaultOutputToParentFlow();
  }

  public Integer getSportId(Event event) {
    return 1;
  }

  @Bean
  public MessageChannel subflowCalculationChannel() {
    return MessageChannels.direct().get();
  }
}

и вот пример одного подпотока:

@Configuration
@AllArgsConstructor
public class CalculatorExample {

  @Bean
  public IntegrationFlow calculateProbabilities(MessageChannel subflowCalculationChannel) {
    return IntegrationFlows.from(subflowCalculationChannel)
        .<Event>handle((p, m) -> p * 2)
        .get();
  }
}

Беда в том, что подпоток пропускает некоторую связь с основным потоком.Я пытался решить эту проблему с помощью defaultOutputToParentFlow () в части маршрутизации, но этого недостаточно.

Answer 1

Начиная с некоторой версии, мы приняли решение привести поведение маршрутизатора Java DSL в соответствие со стандартной конфигурацией с аннотациями или XML.Таким образом, если мы отправляем на маршрутизатор, мы не можем ожидать ответа оттуда.Мы можем продолжить работу только с каналом в качестве выхода из подпотоков.

В вашем случае у вас есть .channel(eventOut) в основном потоке.Итак, все ваши подпотоки маршрутизации должны отвечать именно этому каналу:

    .<Event>handle((p, m) -> corners1H2HCustomBet.getCalculation(p))
    .channel(eventOut)
    .get();

Я думаю, .defaultOutputToParentFlow(); просто ничего не делает для вас, потому что у вас нет отображения по умолчанию.И это уже немного другая история: она не имеет никаких эффектов для других отображений.

Также обратите внимание на этот JavaDoc:

<code>/**
 * Add a subflow as an alternative to a {@link #channelMapping(Object, String)}.
 * {@link #prefix(String)} and {@link #suffix(String)} cannot be used when subflow
 * mappings are used.
 * <p> If subflow should refer to the external {@link IntegrationFlow} bean and
 * there is a requirement to expect reply from there, such a reference should be
 * wrapped with a {@code .gateway()}:
 * <pre class="code">
 * {@code
 *     .subFlowMapping(false, sf -> sf.gateway(evenFlow())))
 * }
 * 

* @param key ключ.* @param subFlow subFlow.* @ вернуть роутер спец.* / public RouterSpecsubFlowMapping (K key, IntegrationFlow subFlow) {

Не относится к вашей конфигурации маршрутизации на основе каналов, но может быть полезно в будущем.

ОБНОВЛЕНИЕ

Вот пример (Kotlin) для subFlowMapping и возврата в основной поток:

    @Bean
    fun splitRouteAggregate() =
            IntegrationFlow { f ->
                f.split()
                        .route<Int, Boolean>({ o -> o % 2 == 0 },
                                { m ->
                                    m.subFlowMapping(true) { sf -> sf.gateway(oddFlow()) }
                                            .subFlowMapping(false) { sf -> sf.gateway(evenFlow()) }
                                })
                        .aggregate()
            }

    @Bean
    fun oddFlow() =
            IntegrationFlow { flow ->
                flow.handle<Any> { _, _ -> "odd" }
            }

    @Bean
    fun evenFlow() =
            IntegrationFlow { flow ->
                flow.handle<Any> { _, _ -> "even" }
            }