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

Revert "Additional changes related to merging 2.2.x to master"

Browse Source 
This reverts commit b9d8c42d81.
This commit is contained in:
Oleg Zhurakousky
2019-02-07 11:51:03 +01:00
parent ac98622ad1
commit 7b7b199b83
12 changed files with 146 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
pom.xml
View File

@@ -28,7 +28,7 @@
<reactor.version>Californium-RELEASE</reactor.version>
<kryo-shaded.version>3.0.3</kryo-shaded.version>
<objenesis.version>2.1</objenesis.version>
<spring-cloud-function.version>2.0.2.BUILD-SNAPSHOT
<spring-cloud-function.version>2.0.0.BUILD-SNAPSHOT
</spring-cloud-function.version>
<maven-checkstyle-plugin.failsOnError>true</maven-checkstyle-plugin.failsOnError>
@@ -273,4 +273,4 @@
</build>
</profile>
</profiles>
</project>
</project>

2
spring-cloud-stream/pom.xml
View File

@@ -108,4 +108,4 @@
</plugins>
</build>
</project>
</project>

5
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java
View File

@@ -217,9 +217,8 @@ public abstract class AbstractBinder<T, C extends ConsumerProperties, P extends
}
else {
rt = StringUtils.hasText(properties.getRetryTemplateName())
? this.consumerBindingRetryTemplates
.get(properties.getRetryTemplateName())
: this.consumerBindingRetryTemplates.values().iterator().next();
? this.consumerBindingRetryTemplates.get(properties.getRetryTemplateName())
: this.consumerBindingRetryTemplates.values().iterator().next();
}
return rt;
}

39
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
View File

@@ -29,6 +29,7 @@ import org.reactivestreams.Publisher;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.DefaultSingletonBeanRegistry;
import org.springframework.cloud.stream.config.ListenerContainerCustomizer;
import org.springframework.cloud.stream.function.IntegrationFlowFunctionSupport;
@@ -555,8 +556,8 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
}
else {
errorChannel = new PublishSubscribeChannel();
((GenericApplicationContext) getApplicationContext()).registerBean(
errorChannelName, SubscribableChannel.class, () -> errorChannel);
((GenericApplicationContext)getApplicationContext()).
registerBean(errorChannelName, SubscribableChannel.class, () -> errorChannel);
}
MessageChannel defaultErrorChannel = null;
if (getApplicationContext()
@@ -570,8 +571,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
errorBridge.setOutputChannel(defaultErrorChannel);
errorChannel.subscribe(errorBridge);
String errorBridgeHandlerName = getErrorBridgeName(destination);
((GenericApplicationContext) getApplicationContext()).registerBean(
errorBridgeHandlerName, BridgeHandler.class, () -> errorBridge);
((GenericApplicationContext)getApplicationContext()).registerBean(errorBridgeHandlerName, BridgeHandler.class, () -> errorBridge);
}
return errorChannel;
}
@@ -613,15 +613,14 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
Object errorChannelObject = getApplicationContext().getBean(errorChannelName);
Assert.isInstanceOf(SubscribableChannel.class, errorChannelObject,
"Error channel '" + errorChannelName
+ "' must be a SubscribableChannel");
"Error channel '" + errorChannelName + "' must be a SubscribableChannel");
errorChannel = (SubscribableChannel) errorChannelObject;
}
else {
errorChannel = new BinderErrorChannel();
((GenericApplicationContext) getApplicationContext()).registerBean(
errorChannelName, SubscribableChannel.class, () -> errorChannel);
((GenericApplicationContext)getApplicationContext())
.registerBean(errorChannelName, SubscribableChannel.class, () -> errorChannel);
}
ErrorMessageSendingRecoverer recoverer;
if (errorMessageStrategy == null) {
@@ -632,10 +631,9 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
errorMessageStrategy);
}
String recovererBeanName = getErrorRecovererName(destination, group,
consumerProperties);
((GenericApplicationContext) getApplicationContext()).registerBean(
recovererBeanName, ErrorMessageSendingRecoverer.class, () -> 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<C extends ConsumerProperties,
if (handler != null) {
if (this.isSubscribable(errorChannel)) {
MessageHandler errorHandler = handler;
((GenericApplicationContext) getApplicationContext()).registerBean(
errorMessageHandlerName, MessageHandler.class,
() -> errorHandler);
((GenericApplicationContext)getApplicationContext()).registerBean(errorMessageHandlerName, MessageHandler.class, () -> errorHandler);
errorChannel.subscribe(handler);
}
else {
@@ -682,10 +678,9 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
errorBridge.setOutputChannel(defaultErrorChannel);
errorChannel.subscribe(errorBridge);
String errorBridgeHandlerName = getErrorBridgeName(destination, group,
consumerProperties);
((GenericApplicationContext) getApplicationContext()).registerBean(
errorBridgeHandlerName, BridgeHandler.class, () -> 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<C extends ConsumerProperties,
private void destroyBean(String beanName) {
if (getApplicationContext().containsBean(beanName)) {
((DefaultSingletonBeanRegistry) getApplicationContext().getBeanFactory())
.destroySingleton(beanName);
((GenericApplicationContext) getApplicationContext())
.removeBeanDefinition(beanName);
((DefaultSingletonBeanRegistry) getApplicationContext().getBeanFactory()).destroySingleton(beanName);
((GenericApplicationContext)getApplicationContext()).removeBeanDefinition(beanName);
}
}

7
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java
View File

@@ -94,7 +94,8 @@ public interface Binding<T> 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<T> 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<T> extends Pausable {
"Binding implementation `" + this.getClass().getName()
+ "` must implement this operation before it is called");
}
}

46
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java
View File

@@ -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<Class<? extends Throwable>, 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;
}
}

4
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java
View File

@@ -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<T, ?, ?> binder = binderProducingContext.getBean(Binder.class);
/*

4
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java
View File

@@ -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<T> implements Binding<T> {
@@ -88,7 +88,7 @@ public class DefaultBinding<T> implements Binding<T> {
public String getBindingName() {
String resolvedName = (this.target instanceof IntegrationObjectSupport)
? ((IntegrationObjectSupport) this.target).getComponentName() : getName();
? ((IntegrationObjectSupport)this.target).getComponentName() : getName();
return resolvedName == null ? getName() : resolvedName;
}

3
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java
View File

@@ -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;

1
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java
View File

@@ -197,6 +197,7 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
return returnMessage;
}
@SuppressWarnings("unchecked")
private <T> Message<O> wrapOutputToMessage(T value, Message<I> originalMessage) {
Message<O> returnMessage = (Message<O>) MessageBuilder.withPayload(value)

62
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/RetryTemplateTests.java
View File

@@ -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<String, RetryTemplate> consumerBindingRetryTemplates = (Map<String, RetryTemplate>) 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<String, RetryTemplate> consumerBindingRetryTemplates = (Map<String, RetryTemplate>) 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();
}
}
}

125
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java
View File

@@ -110,6 +110,66 @@ public class FunctionInvokerTests {
}
}
@EnableAutoConfiguration
@EnableBinding(Processor.class)
public static class ConverterDoesNotProduceCTConfiguration {
@Bean
public Function<String, String> func() {
return x -> x;
}
@StreamMessageConverter
public MessageConverter customConverter() {
return new MessageConverter() {
@Override
public Message<?> toMessage(Object payload, MessageHeaders headers) {
return new GenericMessage<byte[]>(((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<String, String> 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<String, String> 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<String, String> func() {
return x -> x;
}
@StreamMessageConverter
public MessageConverter customConverter() {
return new MessageConverter() {
@Override
public Message<?> toMessage(Object payload, MessageHeaders headers) {
return new GenericMessage<byte[]>(((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 {
}