Настройте подпоток в качестве компонента для PublishSubscribeChannel в DSL интеграции Spring

Question

В соответствии с разделом " Поддержка подпотоков" данного руководства, можно было бы ожидать, что можно будет настроить подпоток очень просто: используя фабрики DSL или лямбды.

// This is modified example from the guide.

@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(
                            // this.subflow1()
                            this.subflow2()

                    ))
            .<Integer>handle((p, h) -> p * 3)
            .channel(c -> c.queue("subscriber3Results"));
}

// Attempt 1:
// Just a copy paste of the logic from above.
// Does not work, java.lang.UnsupportedOperationException

@Bean
public IntegrationFlow subflow1() {
    return f -> f
            .<Integer>handle((p, h) -> p * 2)
            .channel(c -> c.queue("subscriber2Results"));
}

// Attempt 2:
// Using DSL factories.
// Does not work, java.lang.UnsupportedOperationException

@Bean
public IntegrationFlow subflow2() {
    return IntegrationFlows.from(MessageChannels.direct())
            .<Integer>handle((p, h) -> p * 2)
            .channel(c -> c.queue("subscriber2Results"))
            .get();
}

Я получаю это исключение со второй попытки, определенной выше.

Caused by: java.lang.UnsupportedOperationException
at org.springframework.integration.dsl.StandardIntegrationFlow.configure(StandardIntegrationFlow.java:100)
at org.springframework.integration.dsl.PublishSubscribeSpec.subscribe(PublishSubscribeSpec.java:51)

Я попробовал это с spring-boot-starter-integration:2.0.0.M3, Я что-то пропустил? Спасибо за вашу помощь.

Answer 1

Подпотоки не должны быть как бобы. Вы определенно можете извлечь их своими методами, даже private, Но из проводки все вместе вы должны оставить за рамками.

Если вы все еще нуждаетесь в них в качестве bean-компонента, используйте соединение потоков между каналами сообщений.