diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java index 0c5d2b62f..3daee73c7 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java @@ -217,8 +217,9 @@ public abstract class AbstractBinder errorChannel); + ((GenericApplicationContext) getApplicationContext()).registerBean( + errorChannelName, SubscribableChannel.class, () -> errorChannel); } MessageChannel defaultErrorChannel = null; if (getApplicationContext() @@ -571,7 +570,8 @@ public abstract class AbstractMessageChannelBinder errorBridge); + ((GenericApplicationContext) getApplicationContext()).registerBean( + errorBridgeHandlerName, BridgeHandler.class, () -> errorBridge); } return errorChannel; } @@ -613,14 +613,15 @@ public abstract class AbstractMessageChannelBinder errorChannel); + ((GenericApplicationContext) getApplicationContext()).registerBean( + errorChannelName, SubscribableChannel.class, () -> errorChannel); } ErrorMessageSendingRecoverer recoverer; if (errorMessageStrategy == null) { @@ -631,9 +632,10 @@ public abstract class AbstractMessageChannelBinder recoverer); + String recovererBeanName = getErrorRecovererName(destination, group, + consumerProperties); + ((GenericApplicationContext) getApplicationContext()).registerBean( + recovererBeanName, ErrorMessageSendingRecoverer.class, () -> recoverer); MessageHandler handler; if (polled) { handler = getPolledConsumerErrorMessageHandler(destination, group, @@ -660,7 +662,9 @@ public abstract class AbstractMessageChannelBinder errorHandler); + ((GenericApplicationContext) getApplicationContext()).registerBean( + errorMessageHandlerName, MessageHandler.class, + () -> errorHandler); errorChannel.subscribe(handler); } else { @@ -678,9 +682,10 @@ public abstract class AbstractMessageChannelBinder errorBridge); + String errorBridgeHandlerName = getErrorBridgeName(destination, group, + consumerProperties); + ((GenericApplicationContext) getApplicationContext()).registerBean( + errorBridgeHandlerName, BridgeHandler.class, () -> errorBridge); } else { this.logger.warn("The provided errorChannel '" + errorChannelName @@ -766,8 +771,10 @@ public abstract class AbstractMessageChannelBinder extends Pausable { } /** - * Returns the name of the destination for this binding - * + * Returns the name of the destination for this binding. * @return destination name */ default String getName() { @@ -103,8 +102,7 @@ public interface Binding extends Pausable { } /** - * Returns the name of the target for this binding (i.e., channel name) - * + * Returns the name of the target for this binding (i.e., channel name). * @return binding name * * @since 2.2 @@ -132,4 +130,5 @@ public interface Binding extends Pausable { "Binding implementation `" + this.getClass().getName() + "` must implement this operation before it is called"); } + } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java index a858675f2..4b41e161d 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java @@ -72,22 +72,18 @@ public class ConsumerProperties { private int instanceIndex = -1; /** -<<<<<<< HEAD - * The number of attempts to process the message (including the first) in the event of - * processing failures. This is a RetryTemplate configuration which is provided by the - * framework. Default: 3. Set to 1 to disable retry. You can also provide custom - * RetryTemplate in the event you want to take complete control of the RetryTemplate. - * Simply configure it as @Bean inside your application configuration. -======= - * The number of attempts to process the message (including the first) - * in the event of processing failures. This is a RetryTemplate configuration - * which is provided by the framework. - * Default: 3. Set to 1 to disable retry. You can also provide custom RetryTemplate - * in the event you want to take complete control of the RetryTemplate. Simply configure - * it as @Bean inside your application configuration. - * If you need to be binding specific, then you can reference a specific RetryTemplate by name - * with the retry-template configuration. ->>>>>>> 2.2.x + * <<<<<<< HEAD The number of attempts to process the message (including the first) in + * the event of processing failures. This is a RetryTemplate configuration which is + * provided by the framework. Default: 3. Set to 1 to disable retry. You can also + * provide custom RetryTemplate in the event you want to take complete control of the + * RetryTemplate. Simply configure it as @Bean inside your application configuration. + * ======= The number of attempts to process the message (including the first) in the + * event of processing failures. This is a RetryTemplate configuration which is + * provided by the framework. Default: 3. Set to 1 to disable retry. You can also + * provide custom RetryTemplate in the event you want to take complete control of the + * RetryTemplate. Simply configure it as @Bean inside your application configuration. + * If you need to be binding specific, then you can reference a specific RetryTemplate + * by name with the retry-template configuration. >>>>>>> 2.2.x */ private int maxAttempts = 3; @@ -122,18 +118,16 @@ public class ConsumerProperties { private boolean defaultRetryable = true; /** -<<<<<<< HEAD - * A map of Throwable class names in the key and a boolean in the value. Specify those - * exceptions (and subclasses) that will or won't be retried. -======= - * Allows you to further qualify which RetryTemplate to use for a specific consumer binding.. + * <<<<<<< HEAD A map of Throwable class names in the key and a boolean in the value. + * Specify those exceptions (and subclasses) that will or won't be retried. ======= + * Allows you to further qualify which RetryTemplate to use for a specific consumer + * binding.. */ private String retryTemplateName; /** - * A map of Throwable class names in the key and a boolean in the value. - * Specify those exceptions (and subclasses) that will or won't be retried. ->>>>>>> 2.2.x + * A map of Throwable class names in the key and a boolean in the value. Specify those + * exceptions (and subclasses) that will or won't be retried. >>>>>>> 2.2.x */ private Map, Boolean> retryableExceptions = new LinkedHashMap<>(); @@ -172,11 +166,11 @@ public class ConsumerProperties { */ private boolean multiplex; - public String getRetryTemplateName(){ + public String getRetryTemplateName() { return retryTemplateName; } - public void setRetryTemplateName(String retryTemplateName){ + public void setRetryTemplateName(String retryTemplateName) { this.retryTemplateName = retryTemplateName; } @@ -299,4 +293,5 @@ public class ConsumerProperties { public void setAutoStartup(boolean autoStartup) { this.autoStartup = autoStartup; } + } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java index 4ffcb7cdb..39c20b9f5 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java @@ -288,8 +288,8 @@ public class DefaultBinderFactory springApplicationBuilder.environment(binderEnvironment); } - ConfigurableApplicationContext binderProducingContext = - springApplicationBuilder.run(args.toArray(new String[0])); + ConfigurableApplicationContext binderProducingContext = springApplicationBuilder + .run(args.toArray(new String[0])); Binder binder = binderProducingContext.getBean(Binder.class); /* diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java index 64161f4a2..d3d8f65a7 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java @@ -41,7 +41,7 @@ import org.springframework.util.StringUtils; * @see org.springframework.cloud.stream.annotation.EnableBinding */ -@JsonPropertyOrder({"bindingName", "name", "group", "pausable", "state"}) +@JsonPropertyOrder({ "bindingName", "name", "group", "pausable", "state" }) @JsonIgnoreProperties("running") public class DefaultBinding implements Binding { @@ -88,7 +88,7 @@ public class DefaultBinding implements Binding { public String getBindingName() { String resolvedName = (this.target instanceof IntegrationObjectSupport) - ? ((IntegrationObjectSupport)this.target).getComponentName() : getName(); + ? ((IntegrationObjectSupport) this.target).getComponentName() : getName(); return resolvedName == null ? getName() : resolvedName; } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java index cbae6964b..5ff75d5eb 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java @@ -42,8 +42,7 @@ import org.springframework.expression.Expression; public class ProducerProperties { /** - * Signals if this producer needs to be started automatically. - * Default: true + * Signals if this producer needs to be started automatically. Default: true */ private boolean autoStartup = true; diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java index e1d296f9c..337d63929 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java @@ -197,7 +197,6 @@ class FunctionInvoker implements Function>, Flux Message wrapOutputToMessage(T value, Message originalMessage) { Message returnMessage = (Message) MessageBuilder.withPayload(value) diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/RetryTemplateTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/RetryTemplateTests.java index f60be955e..40d5f6bc9 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/RetryTemplateTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/RetryTemplateTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 2018-2019 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. @@ -14,7 +14,6 @@ * limitations under the License. */ - package org.springframework.cloud.stream.config; import java.lang.reflect.Field; @@ -37,11 +36,9 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Import; import org.springframework.retry.support.RetryTemplate; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertTrue; +import static org.assertj.core.api.Assertions.assertThat; /** - * * @author Oleg Zhurakousky * */ @@ -50,50 +47,40 @@ public class RetryTemplateTests { @SuppressWarnings("rawtypes") @Test public void testSingleCustomRetryTemplate() throws Exception { - ApplicationContext context = new SpringApplicationBuilder(SingleCustomRetryTemplateConfiguration.class) - .web(WebApplicationType.NONE) - .run("--spring.jmx.enabled=false"); + ApplicationContext context = new SpringApplicationBuilder( + SingleCustomRetryTemplateConfiguration.class).web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false"); AbstractBinder binder = context.getBean(AbstractBinder.class); Field f = AbstractBinder.class.getDeclaredField("consumerBindingRetryTemplates"); f.setAccessible(true); @SuppressWarnings("unchecked") - Map consumerBindingRetryTemplates = (Map) f.get(binder); - assertTrue(consumerBindingRetryTemplates.size() == 1); - } - - @EnableBinding(Processor.class) - @Import(TestChannelBinderConfiguration.class) - @EnableAutoConfiguration - public static class SingleCustomRetryTemplateConfiguration { - - @StreamRetryTemplate - public RetryTemplate retryTemplate() { - return new RetryTemplate(); - } - - @Bean - public RetryTemplate otherRetryTemplate() { - return new RetryTemplate(); - } + Map consumerBindingRetryTemplates = (Map) f + .get(binder); + assertThat(consumerBindingRetryTemplates).hasSize(1); } @SuppressWarnings("rawtypes") @Test public void testSpecificCustomRetryTemplate() throws Exception { - ApplicationContext context = new SpringApplicationBuilder(SpecificCustomRetryTemplateConfiguration.class) - .web(WebApplicationType.NONE) - .run("--spring.jmx.enabled=false", - "--spring.cloud.stream.bindings.input.consumer.retry-template-name=retryTemplateTwo"); + ApplicationContext context = new SpringApplicationBuilder( + SpecificCustomRetryTemplateConfiguration.class) + .web(WebApplicationType.NONE).run("--spring.jmx.enabled=false", + "--spring.cloud.stream.bindings.input.consumer.retry-template-name=retryTemplateTwo"); - RetryTemplate retryTemplateTwo = context.getBean("retryTemplateTwo", RetryTemplate.class); - BindingServiceProperties bindingServiceProperties = context.getBean(BindingServiceProperties.class); - ConsumerProperties consumerProperties = bindingServiceProperties.getConsumerProperties("input"); + RetryTemplate retryTemplateTwo = context.getBean("retryTemplateTwo", + RetryTemplate.class); + BindingServiceProperties bindingServiceProperties = context + .getBean(BindingServiceProperties.class); + ConsumerProperties consumerProperties = bindingServiceProperties + .getConsumerProperties("input"); AbstractBinder binder = context.getBean(AbstractBinder.class); - Method m = AbstractBinder.class.getDeclaredMethod("buildRetryTemplate", ConsumerProperties.class); + Method m = AbstractBinder.class.getDeclaredMethod("buildRetryTemplate", + ConsumerProperties.class); m.setAccessible(true); - RetryTemplate retryTemplate = (RetryTemplate) m.invoke(binder, consumerProperties); - assertEquals(retryTemplateTwo, retryTemplate); + RetryTemplate retryTemplate = (RetryTemplate) m.invoke(binder, + consumerProperties); + assertThat(retryTemplate).isEqualTo(retryTemplateTwo); } @EnableBinding(Processor.class) @@ -115,5 +102,24 @@ public class RetryTemplateTests { public RetryTemplate otherRetryTemplate() { return new RetryTemplate(); } + } + + @EnableBinding(Processor.class) + @Import(TestChannelBinderConfiguration.class) + @EnableAutoConfiguration + public static class SingleCustomRetryTemplateConfiguration { + + @StreamRetryTemplate + public RetryTemplate retryTemplate() { + return new RetryTemplate(); + } + + @Bean + public RetryTemplate otherRetryTemplate() { + return new RetryTemplate(); + } + + } + } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java index f28f20bcf..fbe8eb629 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java @@ -110,66 +110,6 @@ public class FunctionInvokerTests { } } - @EnableAutoConfiguration - @EnableBinding(Processor.class) - public static class ConverterDoesNotProduceCTConfiguration { - - @Bean - public Function func() { - return x -> x; - } - - @StreamMessageConverter - public MessageConverter customConverter() { - return new MessageConverter() { - - @Override - public Message toMessage(Object payload, MessageHeaders headers) { - return new GenericMessage(((String)payload).getBytes()); - } - - @Override - public Object fromMessage(Message message, Class targetClass) { - String contentType = (String) message.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString(); - if (contentType.equals("foo/bar")) { - return new String((byte[])message.getPayload()); - } - return null; - } - }; - } - } - - @EnableAutoConfiguration - @EnableBinding(Processor.class) - public static class ConverterInjectingCTConfiguration { - - @Bean - public Function func() { - return x -> x; - } - - @StreamMessageConverter - public MessageConverter customConverter() { - return new MessageConverter() { - - @Override - public Message toMessage(Object payload, MessageHeaders headers) { - return MessageBuilder.withPayload(((String)payload).getBytes()).setHeader(MessageHeaders.CONTENT_TYPE, "ping/pong").build(); - } - - @Override - public Object fromMessage(Message message, Class targetClass) { - String contentType = (String) message.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString(); - if (contentType.equals("foo/bar")) { - return new String((byte[])message.getPayload()); - } - return null; - } - }; - } - } - @Test public void testSameMessageTypesAreNotConverted() { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( @@ -364,6 +304,71 @@ public class FunctionInvokerTests { } } + @EnableAutoConfiguration + @EnableBinding(Processor.class) + public static class ConverterDoesNotProduceCTConfiguration { + + @Bean + public Function func() { + return x -> x; + } + + @StreamMessageConverter + public MessageConverter customConverter() { + return new MessageConverter() { + + @Override + public Message toMessage(Object payload, MessageHeaders headers) { + return new GenericMessage(((String) payload).getBytes()); + } + + @Override + public Object fromMessage(Message message, Class targetClass) { + String contentType = (String) message.getHeaders() + .get(MessageHeaders.CONTENT_TYPE).toString(); + if (contentType.equals("foo/bar")) { + return new String((byte[]) message.getPayload()); + } + return null; + } + }; + } + + } + + @EnableAutoConfiguration + @EnableBinding(Processor.class) + public static class ConverterInjectingCTConfiguration { + + @Bean + public Function func() { + return x -> x; + } + + @StreamMessageConverter + public MessageConverter customConverter() { + return new MessageConverter() { + + @Override + public Message toMessage(Object payload, MessageHeaders headers) { + return MessageBuilder.withPayload(((String) payload).getBytes()) + .setHeader(MessageHeaders.CONTENT_TYPE, "ping/pong").build(); + } + + @Override + public Object fromMessage(Message message, Class targetClass) { + String contentType = (String) message.getHeaders() + .get(MessageHeaders.CONTENT_TYPE).toString(); + if (contentType.equals("foo/bar")) { + return new String((byte[]) message.getPayload()); + } + return null; + } + }; + } + + } + @EnableAutoConfiguration public static class MyFunctionsConfiguration {