diff --git a/pom.xml b/pom.xml index 6c30dc7e2..4bbe13d7e 100644 --- a/pom.xml +++ b/pom.xml @@ -28,7 +28,7 @@ Californium-RELEASE 3.0.3 2.1 - 2.0.2.BUILD-SNAPSHOT + 2.0.0.BUILD-SNAPSHOT true @@ -273,4 +273,4 @@ - \ No newline at end of file + diff --git a/spring-cloud-stream/pom.xml b/spring-cloud-stream/pom.xml index 937b02790..b1c5dc0d7 100644 --- a/spring-cloud-stream/pom.xml +++ b/spring-cloud-stream/pom.xml @@ -108,4 +108,4 @@ - \ No newline at end of file + 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 3daee73c7..0c5d2b62f 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,9 +217,8 @@ public abstract class AbstractBinder errorChannel); + ((GenericApplicationContext)getApplicationContext()). + registerBean(errorChannelName, SubscribableChannel.class, () -> errorChannel); } MessageChannel defaultErrorChannel = null; if (getApplicationContext() @@ -570,8 +571,7 @@ public abstract class AbstractMessageChannelBinder errorBridge); + ((GenericApplicationContext)getApplicationContext()).registerBean(errorBridgeHandlerName, BridgeHandler.class, () -> errorBridge); } return errorChannel; } @@ -613,15 +613,14 @@ public abstract class AbstractMessageChannelBinder errorChannel); + ((GenericApplicationContext)getApplicationContext()) + .registerBean(errorChannelName, SubscribableChannel.class, () -> errorChannel); } ErrorMessageSendingRecoverer recoverer; if (errorMessageStrategy == null) { @@ -632,10 +631,9 @@ 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, @@ -662,9 +660,7 @@ public abstract class AbstractMessageChannelBinder errorHandler); + ((GenericApplicationContext)getApplicationContext()).registerBean(errorMessageHandlerName, MessageHandler.class, () -> errorHandler); errorChannel.subscribe(handler); } else { @@ -682,10 +678,9 @@ 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 @@ -771,10 +766,8 @@ 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() { @@ -102,7 +103,8 @@ 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 @@ -130,5 +132,4 @@ 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 0a971978f..a858675f2 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,18 +72,22 @@ 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; @@ -118,15 +122,18 @@ public class ConsumerProperties { private boolean defaultRetryable = true; /** - * - * 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. + * 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<>(); @@ -165,11 +172,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; } @@ -292,5 +299,4 @@ 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 39c20b9f5..4ffcb7cdb 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 d3d8f65a7..64161f4a2 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 5ff75d5eb..cbae6964b 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,7 +42,8 @@ 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 337d63929..e1d296f9c 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,6 +197,7 @@ 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 cad3a7a0c..f60be955e 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 2013-2019 the original author or authors. + * Copyright 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,6 +14,7 @@ * limitations under the License. */ + package org.springframework.cloud.stream.config; import java.lang.reflect.Field; @@ -36,9 +37,11 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Import; import org.springframework.retry.support.RetryTemplate; -import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; /** + * * @author Oleg Zhurakousky * */ @@ -47,40 +50,15 @@ 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); - assertThat(consumerBindingRetryTemplates).hasSize(1); - } - - @Test - @SuppressWarnings("rawtypes") - 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"); - - 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); - m.setAccessible(true); - RetryTemplate retryTemplate = (RetryTemplate) m.invoke(binder, - consumerProperties); - assertThat(retryTemplateTwo).isEqualTo(retryTemplate); + Map consumerBindingRetryTemplates = (Map) f.get(binder); + assertTrue(consumerBindingRetryTemplates.size() == 1); } @EnableBinding(Processor.class) @@ -97,7 +75,25 @@ public class RetryTemplateTests { public RetryTemplate otherRetryTemplate() { return new RetryTemplate(); } + } + @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"); + + 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); + m.setAccessible(true); + RetryTemplate retryTemplate = (RetryTemplate) m.invoke(binder, consumerProperties); + assertEquals(retryTemplateTwo, retryTemplate); } @EnableBinding(Processor.class) @@ -119,7 +115,5 @@ public class RetryTemplateTests { 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 08214c9c2..f28f20bcf 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,6 +110,66 @@ 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( @@ -378,71 +438,6 @@ public class FunctionInvokerTests { } - @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 - @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; - } - }; - } - - } - private static class Foo { }