From 7260eaacabe2c17bd5212992a2e03189c559aa21 Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Thu, 22 Jun 2017 17:27:00 -0400 Subject: [PATCH] Use Reactor version managed by Boot --- ...ageChannelToInputFluxParameterAdapter.java | 4 ++-- .../stream/binding/BindingServiceTests.java | 22 +++++++++---------- 2 files changed, 13 insertions(+), 13 deletions(-) diff --git a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapter.java b/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapter.java index 87ab471e9..28e041bc7 100644 --- a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapter.java +++ b/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapter.java @@ -65,7 +65,7 @@ public class MessageChannelToInputFluxParameterAdapter } }; bindingTarget.subscribe(messageHandler); - emitter.onDispose(() -> bindingTarget.unsubscribe(messageHandler)); + emitter.onCancel(() -> bindingTarget.unsubscribe(messageHandler)); }).publish().autoConnect(); } else { @@ -82,7 +82,7 @@ public class MessageChannelToInputFluxParameterAdapter } }; bindingTarget.subscribe(messageHandler); - emitter.onDispose(() -> bindingTarget.unsubscribe(messageHandler)); + emitter.onCancel(() -> bindingTarget.unsubscribe(messageHandler)); }).publish().autoConnect(); } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java index f49de231a..01cc6a1ad 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java @@ -29,10 +29,9 @@ import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; -import org.springframework.beans.MutablePropertyValues; import org.springframework.beans.factory.NoSuchBeanDefinitionException; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; -import org.springframework.boot.bind.RelaxedDataBinder; +import org.springframework.boot.context.properties.source.MapConfigurationPropertySource; import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.BinderConfiguration; import org.springframework.cloud.stream.binder.BinderType; @@ -311,9 +310,7 @@ public class BindingServiceTests { properties.put("spring.cloud.stream.bindings.input.binder", "mock"); properties.put("spring.cloud.stream.bindings.output.destination", "fooOutput"); properties.put("spring.cloud.stream.bindings.output.binder", "mockError"); - BindingServiceProperties bindingServiceProperties = new BindingServiceProperties(); - RelaxedDataBinder dataBinder = new RelaxedDataBinder(bindingServiceProperties, "spring.cloud.stream"); - dataBinder.bind(new MutablePropertyValues(properties)); + BindingServiceProperties bindingServiceProperties = createBindingServiceProperties(properties); BindingService bindingService = new BindingService(bindingServiceProperties, createMockBinderFactory()); bindingService.bindConsumer(new DirectChannel(), "input"); @@ -334,9 +331,7 @@ public class BindingServiceTests { properties.put("spring.cloud.stream.defaultBinder", "mock1"); properties.put("spring.cloud.stream.binders.mock1.type", "mock"); properties.put("spring.cloud.stream.binders.kafka1.type", "kafka"); - BindingServiceProperties bindingServiceProperties = new BindingServiceProperties(); - RelaxedDataBinder dataBinder = new RelaxedDataBinder(bindingServiceProperties, "spring.cloud.stream"); - dataBinder.bind(new MutablePropertyValues(properties)); + BindingServiceProperties bindingServiceProperties = createBindingServiceProperties(properties); DefaultBinderFactory binderFactory = new BinderFactoryConfiguration() .binderFactory(createMockBinderTypeRegistry(), bindingServiceProperties); BindingService bindingService = new BindingService(bindingServiceProperties, @@ -345,6 +340,13 @@ public class BindingServiceTests { bindingService.bindProducer(new DirectChannel(), "output"); } + private BindingServiceProperties createBindingServiceProperties(HashMap properties) { + BindingServiceProperties bindingServiceProperties = new BindingServiceProperties(); + org.springframework.boot.context.properties.bind.Binder propertiesBinder = new org.springframework.boot.context.properties.bind.Binder(new MapConfigurationPropertySource(properties)); + propertiesBinder.bind("spring.cloud.stream", org.springframework.boot.context.properties.bind.Bindable.ofInstance(bindingServiceProperties)); + return bindingServiceProperties; + } + @Test public void testUnrecognizedBinderDisallowedIfUsed() { HashMap properties = new HashMap<>(); @@ -354,9 +356,7 @@ public class BindingServiceTests { properties.put("spring.cloud.stream.bindings.output.type", "kafka1"); properties.put("spring.cloud.stream.binders.mock1.type", "mock"); properties.put("spring.cloud.stream.binders.kafka1.type", "kafka"); - BindingServiceProperties bindingServiceProperties = new BindingServiceProperties(); - RelaxedDataBinder dataBinder = new RelaxedDataBinder(bindingServiceProperties, "spring.cloud.stream"); - dataBinder.bind(new MutablePropertyValues(properties)); + BindingServiceProperties bindingServiceProperties = createBindingServiceProperties(properties); DefaultBinderFactory binderFactory = new BinderFactoryConfiguration() .binderFactory(createMockBinderTypeRegistry(), bindingServiceProperties); BindingService bindingService = new BindingService(bindingServiceProperties,