diff --git a/pom.xml b/pom.xml
index b74c9a0ac..f0397c0af 100644
--- a/pom.xml
+++ b/pom.xml
@@ -106,7 +106,7 @@
spring-cloud-stream-test-support
spring-cloud-stream-test-support-internal
spring-cloud-stream-integration-tests
-
+ spring-cloud-stream-reactive
spring-cloud-stream-schema
spring-cloud-stream-schema-server
docs
diff --git a/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java b/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java
index 27ec74678..fea760802 100644
--- a/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java
+++ b/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java
@@ -39,8 +39,6 @@ import org.springframework.cloud.stream.binding.StreamListenerMessageHandler;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.config.BindingServiceProperties;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
-import org.springframework.cloud.stream.converter.MessageConverterUtils;
-import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.Lifecycle;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.core.convert.support.DefaultConversionService;
@@ -49,7 +47,6 @@ import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
-import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.converter.SmartMessageConverter;
@@ -84,6 +81,8 @@ public abstract class AbstractBinderTests producerBinding = binder.bindProducer(
- String.format("bad%s0b", getDestinationNameDelimiter()),
- moduleOutputChannel, producerBindingProperties.getProducer());
-
- Binding consumerBinding = binder.bindConsumer(
- String.format("bad%s0b", getDestinationNameDelimiter()), "test-2",
- moduleInputChannel, consumerBindingProperties.getConsumer());
-
- Station station = new Station();
- Message> message = MessageBuilder.withPayload(station).setHeader(
- MessageHeaders.CONTENT_TYPE, MessageConverterUtils.X_JAVA_OBJECT).build();
- moduleInputChannel.subscribe(handler);
- moduleOutputChannel.send(message);
-
- QueueChannel replyChannel = (QueueChannel) handler.getOutputChannel();
-
- Message> replyMessage = replyChannel.receive(5000);
- assertThat(replyMessage.getPayload() instanceof Station).isTrue();
- producerBinding.unbind();
- consumerBinding.unbind();
- }
-
- @SuppressWarnings("rawtypes")
- @Test(expected = MessageDeliveryException.class)
- public void testStreamListenerJavaSerializationNonSerializable() throws Exception {
- Binder binder = getBinder();
-
- BindingProperties producerBindingProperties = createProducerBindingProperties(
- createProducerProperties());
-
- DirectChannel moduleOutputChannel = createBindableChannel("output",
- producerBindingProperties);
-
- BindingProperties consumerBindingProperties = createConsumerBindingProperties(
- createConsumerProperties());
-
- DirectChannel moduleInputChannel = createBindableChannel("input",
- consumerBindingProperties);
-
- Binding producerBinding = binder.bindProducer(
- String.format("bad%s0c", getDestinationNameDelimiter()),
- moduleOutputChannel, producerBindingProperties.getProducer());
-
- Binding consumerBinding = binder.bindConsumer(
- String.format("bad%s0c", getDestinationNameDelimiter()), "test-3",
- moduleInputChannel, consumerBindingProperties.getConsumer());
- try {
- Station station = new Station();
- Message> message = MessageBuilder.withPayload(station)
- .setHeader(MessageHeaders.CONTENT_TYPE,
- MessageConverterUtils.X_JAVA_SERIALIZED_OBJECT)
- .build();
- moduleOutputChannel.send(message);
- }
- finally {
- producerBinding.unbind();
- consumerBinding.unbind();
- }
- }
-
@SuppressWarnings("rawtypes")
@Test
public void testSendJsonReceivePojoWithStreamListener() throws Exception {
@@ -706,8 +625,8 @@ public abstract class AbstractBinderTests handlerClass,
String handlerMethodName, Class>... parameters) throws Exception {
String channelName = "reply_" + System.nanoTime();
- GenericApplicationContext context = new GenericApplicationContext();
- context.getBeanFactory().registerSingleton(channelName, new QueueChannel());
+
+ this.applicationContext.getBeanFactory().registerSingleton(channelName, new QueueChannel());
Method m = ReflectionUtils.findMethod(handlerClass, handlerMethodName,
parameters);
@@ -723,9 +642,9 @@ public abstract class AbstractBinderTests
org.springframework.cloud
spring-cloud-stream-parent
- 2.3.0.BUILD-SNAPSHOT
+ 3.0.0.BUILD-SNAPSHOT
4.0.0