Using an ExpressionEvaluatingRequestHandlerAdvice in a split/aggregate integration flow

Question

I'm looking to implement an integration flow where at some point a split/aggregate writes a bunch of files, returning true or false if an exception occurred in writing one of the files.

It works fine when all files are written without error, I get a list full of trues and can validate my business logic. I've tried to use a ExpressionEvaluatingRequestHandlerAdvice to get a false when writing one the files failed (regardless of the details of the failure), to no avail:

@Bean
public IntegrationFlow mainFlow() {
    return IntegrationFlows.from(...)
            ...
            .split()
            .route(pp -> pp.getPattern == null,
                                mm -> mm.subFlowMapping(true, aSubflow()))
                                        .subFlowMapping(false, anotherSubflow()))                
            // the following line works when I don't use an advice, but does not handle exceptions from above
            //.handle((p, h) -> true)
            .aggregate()
            .>handle((p, h) -> p.stream().allMatch(b -> b))
            ...
           .get();
}

public IntegrationFlow aSubflow() {
    return f -> f.handle(Files.outboundGateway(...),
                    c -> c.advice(fileSystemAdvice()));
}

@Bean
public Advice advice() {
    var advice = new ExpressionEvaluatingRequestHandlerAdvice();
    
    advice.setSuccessChannelName("filesystemSuccess.input");
    advice.setOnSuccessExpressionString("payload + ' was successful'");
    
    advice.setFailureChannelName("filesystemFailure.input");
    advice.setOnFailureExpressionString("payload + ' failed'");
    advice.setTrapException(true);

    return advice;
}

@Bean
public IntegrationFlow success() {
    // logging works fine, but that's not what I want
    //return f -> f.handle(System.out::println);

    return f -> f.handle((p, h) -> true);
}

@Bean
public IntegrationFlow failure() {
    // logging works fine
    //return f -> f.handle(System.out::println);
    return f -> f.handle((p, h) -> false);
}

With this configuration, I get the a DestinationResolutionException no output-channel or replyChannel header available.

What's the correct way to branch the outbound of the success/failure flows back to the split/aggregate flow (just before the aggregate)?

UPDATE:

I also tried to use a gateway as mentioned in the documentation (the splitRouteAggregate() example), but still get the same exception.

UPDATE #2

Here is a minimal failing example: no route, no subflow, just moving a file:

@SpringBootApplication
public class Application {

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

    @Bean
    public IntegrationFlow myFlow() {
        return IntegrationFlows
                .from(Files.inboundAdapter(new File("./")).patternFilter("dummy.txt")
                           .autoCreateDirectory(true).recursive(false)
                           .useWatchService(true).watchEvents(FileReadingMessageSource.WatchEventType.CREATE))
                .handle(Files.outboundGateway(new File("./out"))
                             .deleteSourceFiles(true)
                             .fileExistsMode(FileExistsMode.REPLACE)
                             .autoCreateDirectory(false),
                        c -> c.advice(advice())
                )
                .transform(p -> !Boolean.FALSE.equals(p))
                .log(LoggingHandler.Level.INFO, m -> m.getHeaders().getId() + ": " + m.getPayload())
                .get();
    }
    
    @Bean
    public Advice advice() {
        var advice = new ExpressionEvaluatingRequestHandlerAdvice();
        advice.setOnFailureExpressionString("false");

        return advice;
    }
    
}

Create the out directory at the root of the project, run the project, and then create a dummy.txt at the root of the project, it works fine (though I get two lines of log and not one as I expected).

2022-09-16 23:55:23.678  INFO 21564 --- [   scheduling-1] o.s.integration.handler.LoggingHandler   : 1d253af9-625a-3745-edbb-0438224a9163: true
2022-09-16 23:55:23.700  INFO 21564 --- [   scheduling-1] o.s.integration.handler.LoggingHandler   : 9d2486fc-f776-a5c7-fb21-f16b9ced3094: true

Now, create the out directory at the root of the project, run the project, rmdir the out directory, and then create a dummy.txt at the root of the project, to force an IllegalArgumentException when trying to write the file. I don't get a false payload, the workflow is interrupted by the exception.

    2022-09-17 21:19:00.895 ERROR 14548 --- [   scheduling-1] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessageHandlingException: error occurred in message handler [bean 'myFlow.file:outbound-gateway#0' for component 'myFlow.org.springframework.integration.config.ConsumerEndpointFactoryBean#0'; defined in: 'integrationtest.Application'; from source: 'bean method myFlow']; nested exception is java.lang.IllegalArgumentException: Destination directory [.\out] does not exist., failedMessage=GenericMessage [payload=C:\Users\*****\Dev\ldd\so73745736\.\dummy.txt, headers={file_originalFile=C:\Users\*****\Dev\ldd\so73745736\.\dummy.txt, id=909a6ffe-6598-da35-33ff-90ebb9c68d67, file_name=dummy.txt, file_relativePath=dummy.txt, timestamp=1663442340858}]
        at org.springframework.integration.support.utils.IntegrationUtils.wrapInHandlingExceptionIfNecessary(IntegrationUtils.java:191)
        at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:65)
        at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:115)
        at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:133)
        at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:106)
        at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:72)
        at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:317)
        at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:272)
        at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187)
        at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166)
        at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47)
        at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109)
        at org.springframework.integration.endpoint.SourcePollingChannelAdapter.handleMessage(SourcePollingChannelAdapter.java:196)
        at org.springframework.integration.endpoint.AbstractPollingEndpoint.messageReceived(AbstractPollingEndpoint.java:475)
        at org.springframework.integration.endpoint.AbstractPollingEndpoint.doPoll(AbstractPollingEndpoint.java:461)
        at org.springframework.integration.endpoint.AbstractPollingEndpoint.pollForMessage(AbstractPollingEndpoint.java:413)
        at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$createPoller$4(AbstractPollingEndpoint.java:348)
        at org.springframework.integration.util.ErrorHandlingTaskExecutor.lambda$execute$0(ErrorHandlingTaskExecutor.java:57)
        at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
        at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:55)
        at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$createPoller$5(AbstractPollingEndpoint.java:341)
        at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
        at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:95)
        at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
        at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
        at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
        at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: java.lang.IllegalArgumentException: Destination directory [.\out] does not exist.
        at org.springframework.util.Assert.isTrue(Assert.java:139)
        at org.springframework.integration.file.FileWritingMessageHandler.validateDestinationDirectory(FileWritingMessageHandler.java:496)
        at org.springframework.integration.file.FileWritingMessageHandler.evaluateDestinationDirectoryExpression(FileWritingMessageHandler.java:884)
        at org.springframework.integration.file.FileWritingMessageHandler.handleRequestMessage(FileWritingMessageHandler.java:510)
        at org.springframework.integration.handler.AbstractReplyProducingMessageHandler$AdvisedRequestHandler.handleRequestMessage(AbstractReplyProducingMessageHandler.java:208)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
        at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.base/java.lang.reflect.Method.invoke(Method.java:568)
        at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:344)
        at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:198)
        at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
        at org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice$CallbackImpl.execute(AbstractRequestHandlerAdvice.java:151)
        at org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice.doInvoke(ExpressionEvaluatingRequestHandlerAdvice.java:215)
        at org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice.invoke(AbstractRequestHandlerAdvice.java:67)
        at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
        at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:215)
        at jdk.proxy2/jdk.proxy2.$Proxy46.handleRequestMessage(Unknown Source)
        at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.doInvokeAdvisedRequestHandler(AbstractReplyProducingMessageHandler.java:155)
        at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:139)
        at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:56)
        ... 27 more

Artem Bilan · Answer

The logic of the ExpressionEvaluatingRequestHandlerAdvice is like this:

@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message message) {
    try {
        Object result = callback.execute();
        if (this.onSuccessExpression != null) {
            evaluateSuccessExpression(message);
        }
        return result;
    }
    catch (RuntimeException e) {
        Exception actualException = unwrapExceptionIfNecessary(e);
        if (this.onFailureExpression != null) {
            Object evalResult = evaluateFailureExpression(message, actualException);
            if (this.returnFailureExpressionResult) {
                return evalResult;
            }
        }
        if (!this.trapException) {
            if (e instanceof ThrowableHolderException) { // NOSONAR
                throw (ThrowableHolderException) e;
            }
            else {
                throw new ThrowableHolderException(actualException); // NOSONAR lost stack trace
            }
        }
        return null;
    }
}

So, if callback.execute() is OK, we try to call an onSuccessExpression (if any) and still return the result of the callback.execute(). Looks like in your case it is whatever an HTTP call replies. I don't think we need this onSuccessExpression to return just true. We can do that a bit later with a transform(). Stay tuned.

If callback.execute() fails, we call onFailureExpression, which in your case could be as simple as: setOnFailureExpressionString("false").

Therefore your advice should be configured like this:

@Bean
public Advice advice() {
    var advice = new ExpressionEvaluatingRequestHandlerAdvice();
    advice.setOnFailureExpressionString("false"); 
    advice.setReturnFailureExpressionResult(true);
    return advice;
}

Now, after that Http.outboundGateway() and before an aggregator, you have to add just this:

.transform(p -> !Boolean.FALSE.equals(p))

Using an ExpressionEvaluatingRequestHandlerAdvice in a split/aggregate integration flow

1 Answers