CODEWITHSUNDEEP
javalambdaspring-integration-dsl
pojo-guy
pojo-guy

Reputation: 979

Spring Integration DSL Syntax problem - how to dynamically construct subflows?

I am trying to construct a complex flow in Spring Integration where the sub flows are dynamically defined at runtime. Code that functions perfectly well in the master flow definition fails the compile in the sub flow definition. Since the construct appears identical, it is not obvious what is going on. Any explanation would be appreciated.

Thank you in advance.

The master flow definition is coded something like this:

        StandardIntegrationFlow flow = IntegrationFlows
            .from(setupAdapter,
                    c -> c.poller(Pollers.fixedRate(1000L, TimeUnit.MILLISECONDS).maxMessagesPerPoll(1)))

// This one compiles fine
            .enrichHeaders(h -> h.headerExpression("start", "start\")")
                    .headerExpression("end", "payload[0].get(\"end\")"))

            .split(tableSplitter)
            .enrichHeaders(h -> h.headerExpression("object", "payload[0].get(\"object\")"))
            .channel(c -> c.executor(stepTaskExecutor))
            .routeToRecipients(r -> this.buildRecipientListRouterSpecForRules(r, rules))
            .aggregate()
            .handle(cleanupAdapter).get();

buildRecipientListRouterSpecForRules is defined as:

    private RecipientListRouterSpec buildRecipientListRouterSpecForRules(RecipientListRouterSpec recipientListSpec,
        Collection<RuleMetadata> rules) {
    rules.forEach(
            rule -> recipientListSpec.recipientFlow(getFilterExpression(rule), f -> createFlowDefForRule(f, rule)));

    return recipientListSpec;
}

createFlowDefForRule() is just a switch() wrapper to choose which actual DSL to run for the flow defined by the rule. Here is a sample

    public IntegrationFlowDefinition constructASpecificFlowDef(IntegrationFlowDefinition flowDef, RuleMetadata rule) {

    return flowDef
       // This enrichHeaders element fails to compile,
       // The method headerExpression(String, String) is undefined for the type Object
            .enrichHeaders(h -> h.headerExpression("ALC_operation", "payload[0].get(\"ALC_operation\")"));

 }

Upvotes: 1

Views: 439

Answers (1)

Gary Russell
Gary Russell

Reputation: 174554

In general, it's better to put such explanations in the question text, rather than as comments in the code snippets; I completely missed that comment.

Can you provide a stripped-down (simpler) example (complete class) that exhibits this behavior so we can play with it?

I tried to simplify what you are doing, and this compiles fine and works as expected:

@SpringBootApplication
public class So65010958Application {

    public static void main(String[] args) {
        SpringApplication.run(So65010958Application.class, args);
    }

    @Bean
    IntegrationFlow flow() {
        return IntegrationFlows.from("foo")
                .routeToRecipients(r -> r.recipientFlow("true", f -> buildFlow(f)))
                .get();
    }

    private IntegrationFlowDefinition<?> buildFlow(IntegrationFlowDefinition<?> f) {
        return f.enrichHeaders(h -> h.headerExpression("foo", "'bar'"))
                .channel(MessageChannels.queue("bar"));
    }

    @Bean
    public ApplicationRunner runner(MessageChannel foo, PollableChannel bar) {
        return args -> {
            foo.send(new GenericMessage<>("foo"));
            System.out.println(bar.receive(0));
        };
    }

}

GenericMessage [payload=foo, headers={foo=bar, id=d526b8fb-c6f8-7731-b1ad-e68e326fcc00, timestamp=1606333567749}]

So, I must be missing something.

Upvotes: 1

Related Questions