Spring Integration настраивает существующий поток

Question

Если у нас есть что-то вроде этого:

private void createFlow(String pathOfEndpoint) {

        IntegrationFlow integrationFlow = IntegrationFlows
                                                    .from(confGateway(pathOfEndpoint))
                                                    .<Map<String,String>>handle((p, h) -> {
                                                        return doSomething(p);
                                                    })
                                                    .get();

        getIntegrationFlowContext().registration(integrationFlow).id(pathOfEndpoint).register();
    }

И confGateway - это:

HttpRequestHandlingMessagingGateway gateway = new HttpRequestHandlingMessagingGateway(true);
...
return gateway;

Итак, мы создали поток и зарегистрировались. У меня вопрос, есть ли способ настроить этот поток больше во время выполнения, например, через некоторое время добавить еще .handle((p,h) -> .... или, если необходимо, добавить .transform(...). Я пробовал с integrationFlow.configure(....), но я видел, что метод в StandardIntegrationFlow таков:

@Override
    public void configure(IntegrationFlowDefinition<?> flow) {
        throw new UnsupportedOperationException();
    }

Можно ли как-нибудь позже добавить еще один .handle(...) или аналогичный? Если нет, есть ли другой способ, которым я могу манипулировать существующим потоком? может быть subFlow? Я хотел бы поблагодарить Артема Билана за отличную поддержку и примеры для интеграции в Spring. Продолжайте в том же духе.

Answer 1

Непонятно, зачем создавать IntegrationFlow во время выполнения, а затем решил изменить его позже ... Однако это не имеет значения. Вы должны понять для себя, что IntegrationFlow это просто логический компонент конфигурации. Он ничего не делает во время выполнения, во время обработки сообщений. Что я хочу сказать, что во время выполнения у нас есть группа активных компонентов, которые ничего не знают о IntegrationFlow, создавшем их. Они уже связаны друг с другом с использованием принципов EIP - каналов, конечных точек, программ опроса и т. Д. c.

Конечно, эти компоненты являются гибкими и могут быть настроены во время выполнения, но определенно не IntegrationFlow создал их. Я имею в виду, что вы можете подписаться и отписаться на / из каналов во время выполнения, используя их контракт. Или вы можете добавить (end remove - stop() it) конечную точку опроса в существующую QueueChannel. Все остальные компоненты не будут затронуты. Принцип слабой связи является одним из основных пунктов решения EIP.

Итак, что вы спрашиваете о Dynami c .handle() или .transform() больше о подписке на новые потребители к существующим каналам. Поскольку это довольно сложно сделать самостоятельно (вам нужно немного углубиться в основы Spring), вы все равно можете использовать упомянутый IntegrationFlowContext.registration()). Вам нужно иметь имя канала из своего исходного потока в конце и использовать этот канал (или его имя) для создания нового IntegrationFlow с необходимыми .handle() или .transform() logi c:

IntegrationFlow integrationFlow = 
                    IntegrationFlows
                             .from(confGateway(pathOfEndpoint))
                             .<Map<String,String>>handle((p, h) -> {
                                            return doSomething(p);
                             })
                             .channel("mainChannel")
                             .get();

 ... 

 IntegrationFlow otherFlow = 
                    IntegrationFlows
                             .from("mainChannel")
                             .transform(...)
                             .get();

Answer 2

Нет; вы не можете изменить поток во время выполнения; однако вы можете использовать dynamici c и потоки интеграции времени выполнения , удалить поток и заменить его новым.