Как создавать потоки Spring Integration через DSL

Question

Я пытаюсь выяснить, как разделить потоки интеграции пружины на несколько подпотоков и собрать их вместе.В конечном итоге я пытаюсь изложить шаблон, в котором я могу создать модули подпотоков, которые можно объединить для общих рецептов интеграции.

Этот тестовый пример представляет собой минимальный пример попытки (но неуспешного) подключениявместе подпотоки с помощью DSL IntegrationFlow API:

@RunWith(SpringRunner.class)
@SpringBootTest(classes = { ComposedIntegrationFlowTest.class })
@SpringIntegrationTest
@EnableIntegration
public class ComposedIntegrationFlowTest {

    @Test
    public void test() {
        MessageChannel beginningChannel = MessageChannels.direct("beginning").get();

        IntegrationFlow middleFlow = f -> f
            .transform("From middleFlow: "::concat)
            .transform(String.class, String::toUpperCase);

        IntegrationFlow endFlow = f -> f
            .handle((p, h) -> "From endFlow: " + p);

        StandardIntegrationFlow composedFlow = IntegrationFlows
            .from(beginningChannel)
            .gateway(middleFlow)
            .gateway(endFlow)
            .get();

        composedFlow.start();

        beginningChannel.send(MessageBuilder.withPayload("hello!").build());
    }
}

Пытаясь описать выше, я получаю:

org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'beginning'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload=hello!, headers={id=2b1de253-a822-42ba-cd85-009b83a644eb, timestamp=1537890950879}], failedMessage=GenericMessage [payload=hello!, headers={id=2b1de253-a822-42ba-cd85-009b83a644eb, timestamp=1537890950879}]

Как собрать воедино эти подпотоки?Есть ли лучший API для создания такого рода композиции?Правильно ли построен этот интеграционный тест?

Answer 1

medFlow.start ();

Этого недостаточно;вам нужно зарегистрировать динамические потоки с помощью IntegrationFlowContext, чтобы все поддерживающие компоненты регистрировались в контексте приложения.