SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove error.destination property

Browse Source 
Remove the ability to bind global error channel directly to a broker destination through error.destination property.
The preferred way to consume error messages in a brokder destination is through the binder specific DLQ mechanism.
Global error channel may still be consumed by a ServiceActivator or StreamListener, but it is not bound to a broker destination.

Remove the internal bridge channel from the global error channel that is used for content type conversion.
Remove tests associated with this feature.
Deprecate class `SingleBindingTargetBindable` as this is not used by the framework any longer.
Docs polishing.

Resolves #1398
Resolves #1401
Resolves #1269
This commit is contained in:
Soby Chacko
2018-06-26 11:58:35 -04:00
committed by Oleg Zhurakousky
parent 9626b5881c
commit a82c906913
6 changed files with 3 additions and 221 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc
View File

@@ -737,18 +737,6 @@ public void error(Message<?> message) {
This may be a convenient option if error handling logic is the same regardless of which handler produced the error.
Also, error messages sent to the `errorChannel` can be published to the specific destination at the broker by configuring a binding named `error` for the outbound target.
This option provides a mechanism to automatically send error messages to another application bound to that destination or for later retrieval (for example, audit).
For example, to publish error messages to a broker destination named `myErrors`, set the following property:
[source,text]
----
spring.cloud.stream.bindings.error.destination=myErrors.
----
NOTE: The ability to bridge global error channel to a broker destination essentially provides a mechanism which connects
the _application-level_ error handling with the _system-level_ error handling.
===== System Error Handling
System-level error handling implies that the errors are communicated back to the messaging system and, given that not every messaging system

82
spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/ErrorChannelTests.java
View File

@@ -1,82 +0,0 @@
/*
* Copyright 2016 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.config;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.cloud.stream.test.binder.TestSupportBinder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.PropertySource;
import org.springframework.integration.annotation.InboundChannelAdapter;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.core.MessageSource;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.Assert;
/**
* @author Ilayaperumal Gopinathan
*/
@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest(classes = ErrorChannelTests.TestSource.class)
public class ErrorChannelTests {
@Autowired
@Qualifier(BindingServiceConfiguration.ERROR_BRIDGE_CHANNEL)
private MessageChannel errorBridgeChannel;
@Autowired
private BinderFactory binderFactory;
@Test
public void testErrorChannelBinding() throws Exception {
Message<?> message = ((TestSupportBinder) binderFactory.getBinder(null, MessageChannel.class))
.messageCollector().forChannel(errorBridgeChannel).poll(10, TimeUnit.SECONDS);
Assert.isTrue(message instanceof ErrorMessage, "Message should be an instance of ErrorMessage");
Assert.isTrue(message.getPayload() instanceof MessagingException, "Message payload should be an instance" +
"of MessagingException");
Assert.isTrue(message.getPayload().toString()
.equals("org.springframework.messaging.MessagingException: test"), "Text did not match");
}
@EnableBinding(Source.class)
@EnableAutoConfiguration
@PropertySource("classpath:/org/springframework/cloud/stream/config/errorchannel/source-channel.properties")
public static class TestSource {
@Bean
@InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "5000", maxMessagesPerPoll = "1"))
public MessageSource<String> timerMessageSource() {
return () -> {
throw new MessagingException("test");
};
}
}
}

2
spring-cloud-stream-integration-tests/src/test/resources/org/springframework/cloud/stream/config/errorchannel/source-channel.properties
View File

@@ -1,2 +0,0 @@
spring.cloud.stream.bindings.output.destination=source-output
spring.cloud.stream.bindings.error.destination=errorchannel-test

3
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SingleBindingTargetBindable.java
View File

@@ -31,7 +31,10 @@ import org.springframework.cloud.stream.binder.Binding;
*
* @author Ilayaperumal Gopinathan
* @author Marius Bogoevici
*
* @deprecated This class is no longer used by the framework and maybe removed in a future release.
*/
@Deprecated
public class SingleBindingTargetBindable<T> implements Bindable {
private final String name;

32
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java
View File

@@ -24,7 +24,6 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.binding.AbstractBindingTargetFactory;
@@ -40,7 +39,6 @@ import org.springframework.cloud.stream.binding.MessageChannelStreamListenerResu
import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
import org.springframework.cloud.stream.binding.MessageSourceBindingTargetFactory;
import org.springframework.cloud.stream.binding.OutputBindingLifecycle;
import org.springframework.cloud.stream.binding.SingleBindingTargetBindable;
import org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor;
import org.springframework.cloud.stream.binding.SubscribableChannelBindingTargetFactory;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
@@ -52,17 +50,13 @@ import org.springframework.context.annotation.DependsOn;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.Role;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.config.GlobalChannelInterceptorProcessor;
import org.springframework.integration.config.HandlerMethodArgumentResolversHolder;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.BridgeHandler;
import org.springframework.integration.router.AbstractMappingMessageRouter;
import org.springframework.lang.Nullable;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.core.DestinationResolver;
import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory;
@@ -90,11 +84,6 @@ public class BindingServiceConfiguration {
public static final String STREAM_LISTENER_ANNOTATION_BEAN_POST_PROCESSOR_NAME =
"streamListenerAnnotationBeanPostProcessor";
public static final String ERROR_BRIDGE_CHANNEL = "errorBridgeChannel";
private static final String ERROR_KEY_NAME = "error";
@Bean
public MessageChannelStreamListenerResultAdapter messageChannelStreamListenerResultAdapter() {
return new MessageChannelStreamListenerResultAdapter();
@@ -180,32 +169,11 @@ public class BindingServiceConfiguration {
callback, globalChannelInterceptorProcessor);
}
@Bean
@ConditionalOnProperty("spring.cloud.stream.bindings." + ERROR_KEY_NAME + ".destination")
public MessageChannel errorBridgeChannel(
@Qualifier(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME) PublishSubscribeChannel errorChannel) {
SubscribableChannel errorBridgeChannel = new DirectChannel();
BridgeHandler handler = new BridgeHandler();
handler.setOutputChannel(errorBridgeChannel);
errorChannel.subscribe(handler);
return errorBridgeChannel;
}
@Bean
@ConditionalOnProperty("spring.cloud.stream.bindings." + ERROR_KEY_NAME + ".destination")
public SingleBindingTargetBindable<MessageChannel> errorBridgeChannelBindable(
@Qualifier(ERROR_BRIDGE_CHANNEL) MessageChannel errorBridgeChannel,
CompositeMessageChannelConfigurer compositeMessageChannelConfigurer) {
compositeMessageChannelConfigurer.configureOutputChannel(errorBridgeChannel, ERROR_KEY_NAME);
return new SingleBindingTargetBindable<>(ERROR_KEY_NAME, errorBridgeChannel);
}
@Bean
public DynamicDestinationsBindable dynamicDestinationsBindable() {
return new DynamicDestinationsBindable();
}
@SuppressWarnings("deprecation")
@Bean
@ConditionalOnMissingBean

93
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ErrorBindingTests.java
View File

@@ -16,29 +16,21 @@
package org.springframework.cloud.stream.binder;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.config.BindingServiceConfiguration;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Import;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.GenericMessage;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.ArgumentMatchers.same;
/**
* @author Marius Bogoevici
@@ -61,95 +53,10 @@ public class ErrorBindingTests {
applicationContext.close();
}
@SuppressWarnings({"rawtypes", "unchecked"})
@Test
public void testErrorChannelBoundIfConfigured() {
ConfigurableApplicationContext applicationContext = SpringApplication.run(TestProcessor.class,
"--spring.cloud.stream.bindings.error.destination=foo", "--server.port=0");
BinderFactory binderFactory = applicationContext.getBean(BinderFactory.class, MessageChannel.class);
Binder binder = binderFactory.getBinder(null, MessageChannel.class);
MessageChannel errorChannel = applicationContext.getBean(BindingServiceConfiguration.ERROR_BRIDGE_CHANNEL,
MessageChannel.class);
Mockito.verify(binder).bindConsumer(eq("input"), isNull(), any(MessageChannel.class),
any(ConsumerProperties.class));
Mockito.verify(binder).bindProducer(eq("output"), any(MessageChannel.class), any(ProducerProperties.class));
Mockito.verify(binder).bindProducer(eq("foo"), same(errorChannel), any(ProducerProperties.class));
Mockito.verifyNoMoreInteractions(binder);
applicationContext.close();
}
@Test
public void testErrorChannelIsBoundWithCorrectContentTypeConverter() {
final AtomicBoolean received = new AtomicBoolean(false);
ConfigurableApplicationContext applicationContext = SpringApplication.run(TestProcessor.class,
"--spring.cloud.stream.bindings.error.destination=foo",
"--spring.cloud.stream.bindings.error.content-type=application/json",
"--server.port=0");
MessageChannel errorChannel = applicationContext.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME,
MessageChannel.class);
MessageChannel errorBridgeChannel = applicationContext.getBean(BindingServiceConfiguration.ERROR_BRIDGE_CHANNEL,
MessageChannel.class);
((SubscribableChannel) errorBridgeChannel).subscribe(message -> {
assertThat(new String((byte[]) message.getPayload())).isEqualTo("{\"foo\":\"bar\"}");
received.set(true);
});
Foo foo = new Foo();
foo.setFoo("bar");
errorChannel.send(new GenericMessage<>(foo));
assertThat(received.get()).isTrue();
applicationContext.close();
}
@Test
public void testErrorChannelForExceptionWhenContentTypeIsSet() {
final AtomicBoolean received = new AtomicBoolean(false);
ConfigurableApplicationContext applicationContext = SpringApplication.run(TestProcessor.class,
"--spring.cloud.stream.bindings.error.destination=foo",
"--spring.cloud.stream.bindings.error.content-type=application/json",
"--server.port=0");
MessageChannel errorChannel = applicationContext.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME,
MessageChannel.class);
MessageChannel errorBridgeChannel = applicationContext.getBean(BindingServiceConfiguration.ERROR_BRIDGE_CHANNEL,
MessageChannel.class);
((SubscribableChannel) errorBridgeChannel).subscribe(message -> {
String payload = new String((byte[]) message.getPayload());
assertThat(payload.contains("cause")).isTrue();
assertThat(payload.contains("stackTrace")).isTrue();
assertThat(payload.contains("throwing exception")).isTrue();
received.set(true);
});
errorChannel.send(new GenericMessage<>(new Exception("throwing exception")));
assertThat(received.get()).isTrue();
applicationContext.close();
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
@Import(MockBinderRegistryConfiguration.class)
public static class TestProcessor {
}
private class Foo {
String foo;
@SuppressWarnings("unused") // used json ser/deser
public String getFoo() {
return foo;
}
public void setFoo(String foo) {
this.foo = foo;
}
}
}