GH-2799: Fix AnnGateProxyFB for empty errChannel

Fixes https://github.com/spring-projects/spring-integration/issues/2799

**Cherry-pick to 5.1.x & 5.0.x**

When `AnnotationGatewayProxyFactoryBean` is used for the one-way (`void`)
POJI method invocation and downstream processing is based on an
`ExecutorChannel`, the default `errorChannel` value from the annotation
is resolved to the empty string with is set to the target gateway proxy
and can not be resolved to the target bean eventually in the
`MessagePublishingErrorHandler` in case of exception

* Check `errorChannel` attribute for empty string and don't set it into
the `errorChannelName`.
This way the `MessagePublishingErrorHandler` will resolve to the global
`errorChannel` as expected.
This commit is contained in:
Artem Bilan
2019-03-11 13:25:33 -04:00
committed by Gary Russell
parent a42411f2d6
commit 3f3d98c2d4
2 changed files with 41 additions and 3 deletions

View File

@@ -100,7 +100,9 @@ public class AnnotationGatewayProxyFactoryBean extends GatewayProxyFactoryBean {
}
String errorChannel = beanFactory.resolveEmbeddedValue(this.gatewayAttributes.getString("errorChannel"));
setErrorChannelName(errorChannel);
if (StringUtils.hasText(errorChannel)) {
setErrorChannelName(errorChannel);
}
String asyncExecutor = beanFactory.resolveEmbeddedValue(this.gatewayAttributes.getString("asyncExecutor"));
if (asyncExecutor == null || AnnotationConstants.NULL.equals(asyncExecutor)) {
@@ -146,11 +148,15 @@ public class AnnotationGatewayProxyFactoryBean extends GatewayProxyFactoryBean {
String defaultRequestTimeout =
beanFactory.resolveEmbeddedValue(this.gatewayAttributes.getString("defaultRequestTimeout"));
setDefaultRequestTimeout(Long.parseLong(defaultRequestTimeout));
if (StringUtils.hasText(defaultRequestTimeout)) {
setDefaultRequestTimeout(Long.parseLong(defaultRequestTimeout));
}
String defaultReplyTimeout =
beanFactory.resolveEmbeddedValue(this.gatewayAttributes.getString("defaultReplyTimeout"));
setDefaultReplyTimeout(Long.parseLong(defaultReplyTimeout));
if (StringUtils.hasText(defaultReplyTimeout)) {
setDefaultReplyTimeout(Long.parseLong(defaultReplyTimeout));
}
super.onInit();
}

View File

@@ -33,6 +33,7 @@ import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
@@ -53,6 +54,7 @@ import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.MessageDispatchingException;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.MessagingGateway;
@@ -501,6 +503,26 @@ public class IntegrationFlowTests {
this.flow2WithPrototypeHandlerConsumer.getHandler());
}
@Autowired
@Qualifier("globalErrorChannelResolutionFunction")
private Consumer<String> globalErrorChannelResolutionGateway;
@Autowired
SubscribableChannel errorChannel;
@Test
public void testGlobalErrorChannelResolutionFlow() throws InterruptedException {
CountDownLatch errorMessageLatch = new CountDownLatch(1);
MessageHandler errorMessageHandler = m -> errorMessageLatch.countDown();
this.errorChannel.subscribe(errorMessageHandler);
this.globalErrorChannelResolutionGateway.accept("foo");
assertThat(errorMessageLatch.await(10, TimeUnit.SECONDS)).isTrue();
this.errorChannel.unsubscribe(errorMessageHandler);
}
@MessagingGateway
public interface ControlBusGateway {
@@ -890,6 +912,16 @@ public class IntegrationFlowTests {
return f -> f.handle(handler, e -> e.id("flow2WithPrototypeHandlerConsumer"));
}
@Bean
public IntegrationFlow globalErrorChannelResolutionFlow(@Qualifier("taskScheduler") TaskExecutor taskExecutor) {
return IntegrationFlows.from(Consumer.class, "globalErrorChannelResolutionFunction")
.channel(c -> c.executor(taskExecutor))
.handle((GenericHandler<?>) (p, h) -> {
throw new RuntimeException("intentional");
})
.get();
}
}
@Service