From a7fdf6dbb2c87319024e29898484e299bf4f4328 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Sat, 28 Oct 2017 18:02:42 -0400 Subject: [PATCH] GH-1106 Addressed backward/forward contentType compatibility issues Fixes #1106 - This PR builds on the previous PR with commit hash 6c259be6... (pr/1112) - Added support in AbstractBinderTests to create bindable channel based on determining channel input type based on it's name (i.e., *input*) - Added LegacyContentTypeHeaderInterceptor to TestSupportBinder to restor the previous behavior of MessageCollector for cases where Message's payload content type is a variant of 'text'. - Restored tests that use MessageCollector to depend on proper payload type - Added 'deserialize' routine back to MessageSerializationUtils - Polished MessageConverterConfigurer.LegacyContentTypeHeaderInterceptor to remove conditional original-content-type header logic - Removed default contentType from BindingProperties - Restored tests that use MessageCollector to depend on proper payload type - polishing - fixed Kryo/Java serialization - Fixed ser/de for "application/json" and contentType equals - Fixed of ser/de of JSON strings to ensure that Strings are not re-quoted - Fixed how we comparing contentTypes - more polishing - Fixed NPE in MessageSerializationUtils - Make bindings and consumer groups in new tests added to AbstractBinderTests mutually exclusive --- .../stream/binder/AbstractBinderTests.java | 511 +++++++++++++++++- .../stream/binder/AbstractTestBinder.java | 1 - .../cloud/stream/binder/SerializableFoo.java | 34 ++ .../ContentTypeOutboundSourceTests.java | 10 +- .../config/CustomHeaderPropagationTests.java | 5 +- .../config/DefaultHeaderPropagationTests.java | 5 +- ...ionWithApplicationProvidedHeaderTests.java | 13 +- .../DeserializeJSONToJavaTypeTests.java | 13 +- .../InboundJsonToTupleConversionTest.java | 5 +- .../stream/config/LegacyContentTypeTests.java | 7 +- ...ListenerAnnotatedMethodArgumentsTests.java | 8 +- .../StreamListenerHandlerBeanTests.java | 9 +- .../StreamListenerHandlerMethodTests.java | 34 +- .../StreamListenerMessageArgumentTests.java | 9 +- ...stenerMethodReturnWithConversionTests.java | 11 +- ...mListenerMethodWithReturnMessageTests.java | 9 +- ...eamListenerMethodWithReturnValueTests.java | 9 +- ...enerWithAnnotatedInputOutputArgsTests.java | 9 +- .../StreamListenerWithConditionsTest.java | 3 - .../config/TextPlainConversionTest.java | 13 +- .../config/TextPlainToJsonConversionTest.java | 8 +- .../aggregate/AggregateApplicationTests.java | 6 +- .../config/contentType/ContentTypeTests.java | 43 +- .../reactive/StreamEmitterBasicTests.java | 24 +- ...icFluxInputOutputArgsWithMessageTests.java | 8 +- ...mListenerReactiveInputOutputArgsTests.java | 11 +- ...activeInputOutputArgsWithMessageTests.java | 11 +- ...utOutputArgsWithSenderAndFailureTests.java | 12 +- ...eactiveInputOutputArgsWithSenderTests.java | 11 +- ...enerReactiveMethodWithReturnTypeTests.java | 11 +- ...istenerReactiveReturnWithFailureTests.java | 12 +- ...istenerReactiveReturnWithMessageTests.java | 11 +- ...amListenerReactiveReturnWithPojoTests.java | 10 +- ...rdFluxInputOutputArgsWithMessageTests.java | 9 +- .../stream/test/binder/TestSupportBinder.java | 7 +- .../aggregate/bean/AggregateWithBeanTest.java | 5 +- .../aggregate/main/AggregateWithMainTest.java | 5 +- .../AutoconfigurationDisabledTest.java | 5 +- .../stream/test/example/ExampleTest.java | 10 +- .../binder/MessageSerializationUtils.java | 32 +- .../binding/MessageConverterConfigurer.java | 97 +++- ...amListenerAnnotationBeanPostProcessor.java | 9 +- .../stream/config/BindingProperties.java | 7 +- .../CompositeMessageConverterFactory.java | 37 +- ...CustomJackson2MappingMessageConverter.java | 54 -- .../JavaSerializationMessageConverter.java | 8 +- .../MessageConverterConfigurerTests.java | 13 +- ...mMappingJackson2MessageConverterTests.java | 63 --- 48 files changed, 875 insertions(+), 382 deletions(-) create mode 100644 spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/SerializableFoo.java delete mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CustomJackson2MappingMessageConverter.java delete mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/converter/CustomMappingJackson2MessageConverterTests.java 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 b6498ee8e..150ba703c 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 @@ -1,5 +1,5 @@ /* - * Copyright 2013-2016 the original author or authors. + * Copyright 2013-2017 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. @@ -16,17 +16,29 @@ package org.springframework.cloud.stream.binder; +import java.io.Serializable; +import java.lang.reflect.Constructor; +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.junit.After; import org.junit.Test; +import org.springframework.cloud.stream.binder.AbstractBinderTests.Station.Readings; import org.springframework.cloud.stream.binding.MessageConverterConfigurer; +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; @@ -36,11 +48,22 @@ 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.MessageHandler; +import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.handler.annotation.support.PayloadArgumentResolver; +import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite; +import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; +import org.springframework.util.Assert; +import org.springframework.util.MimeType; import org.springframework.util.MimeTypeUtils; +import org.springframework.util.ReflectionUtils; import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; /** * @author Gary Russell @@ -48,6 +71,7 @@ import static org.assertj.core.api.Assertions.assertThat; * @author David Turanski * @author Mark Fisher * @author Marius Bogoevici + * @author Oleg Zhurakousky */ @SuppressWarnings("unchecked") public abstract class AbstractBinderTests, CP, PP>, CP extends ConsumerProperties, PP extends ProducerProperties> { @@ -87,6 +111,7 @@ public abstract class AbstractBinderTests foo0ProducerBinding = binder.bindProducer("foo.0", @@ -121,36 +146,156 @@ public abstract class AbstractBinderTests producerBinding = binder.bindProducer("foo.0", - moduleOutputChannel, outputBindingProperties.getProducer()); - Binding consumerBinding = binder.bindConsumer("foo.0", - "testSendAndReceive", moduleInputChannel, createConsumerProperties()); - // Bypass conversion we are only testing sendReceive - Message message = MessageBuilder.withPayload("foo".getBytes()) - .setHeader(MessageHeaders.CONTENT_TYPE, - MimeTypeUtils.APPLICATION_OCTET_STREAM) + BindingProperties outputBindingProperties = createProducerBindingProperties(createProducerProperties()); + DirectChannel moduleOutputChannel = createBindableChannel("output", outputBindingProperties); + + BindingProperties inputBindingProperties = createConsumerBindingProperties(createConsumerProperties()); + DirectChannel moduleInputChannel = createBindableChannel("input", inputBindingProperties); + + Binding producerBinding = binder.bindProducer("foo.0", moduleOutputChannel, + outputBindingProperties.getProducer()); + Binding consumerBinding = binder.bindConsumer("foo.0", "testSendAndReceive", moduleInputChannel, + createConsumerProperties()); + Message message = MessageBuilder.withPayload("foo").setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar") .build(); // Let the consumer actually bind to the producer before sending a msg binderBindUnbindLatency(); + + CountDownLatch latch = new CountDownLatch(1); + AtomicReference> inboundMessageRef = new AtomicReference>(); + moduleInputChannel.subscribe(new MessageHandler() { + + @Override + public void handleMessage(Message message) throws MessagingException { + try { + inboundMessageRef.set((Message) message); + } + finally { + latch.countDown(); + } + } + }); + moduleOutputChannel.send(message); - Message inbound = receive(moduleInputChannel); - assertThat(inbound).isNotNull(); - assertThat(inbound.getPayload()).isEqualTo("foo".getBytes()); - assertThat(inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE)) - .isEqualTo(MimeTypeUtils.APPLICATION_OCTET_STREAM_VALUE); + Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message"); + + assertThat(inboundMessageRef.get().getPayload()).isEqualTo("foo"); + assertThat(inboundMessageRef.get().getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull(); + assertThat(inboundMessageRef.get().getHeaders().get(MessageHeaders.CONTENT_TYPE).toString()).isEqualTo("foo/bar"); + producerBinding.unbind(); + consumerBinding.unbind(); + } + + private class Foo { + private String name; + + @SuppressWarnings("unused") + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + } + + @SuppressWarnings("rawtypes") + @Test + public void testSendAndReceiveKryo() throws Exception { + Binder binder = getBinder(); + BindingProperties outputBindingProperties = createProducerBindingProperties(createProducerProperties()); + DirectChannel moduleOutputChannel = createBindableChannel("output", outputBindingProperties); + + BindingProperties inputBindingProperties = createConsumerBindingProperties(createConsumerProperties()); + DirectChannel moduleInputChannel = createBindableChannel("input", inputBindingProperties); + + Binding producerBinding = binder.bindProducer("foo.0x", moduleOutputChannel, + outputBindingProperties.getProducer()); + Binding consumerBinding = binder.bindConsumer("foo.0x", "testSendAndReceiveKryo", moduleInputChannel, + createConsumerProperties()); + Foo foo = new Foo(); + foo.setName("Bill"); + Message message = MessageBuilder.withPayload(foo).setHeader(MessageHeaders.CONTENT_TYPE, MessageConverterUtils.X_JAVA_OBJECT) + .build(); + // Let the consumer actually bind to the producer before sending a msg + binderBindUnbindLatency(); + + CountDownLatch latch = new CountDownLatch(1); + AtomicReference> inboundMessageRef = new AtomicReference>(); + moduleInputChannel.subscribe(new MessageHandler() { + + @Override + public void handleMessage(Message message) throws MessagingException { + try { + inboundMessageRef.set((Message) message); + } + finally { + latch.countDown(); + } + } + }); + + moduleOutputChannel.send(message); + Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message"); + + assertThat(inboundMessageRef.get().getPayload()).isInstanceOf(Foo.class); + assertThat(inboundMessageRef.get().getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull(); + assertTrue(equalTypeAndSubType((MimeType)inboundMessageRef.get().getHeaders().get(MessageHeaders.CONTENT_TYPE), MessageConverterUtils.X_JAVA_OBJECT)); + producerBinding.unbind(); + consumerBinding.unbind(); + } + + @SuppressWarnings("rawtypes") + @Test + public void testSendAndReceiveJavaSerialization() throws Exception { + Binder binder = getBinder(); + BindingProperties outputBindingProperties = createProducerBindingProperties(createProducerProperties()); + DirectChannel moduleOutputChannel = createBindableChannel("output", outputBindingProperties); + + BindingProperties inputBindingProperties = createConsumerBindingProperties(createConsumerProperties()); + DirectChannel moduleInputChannel = createBindableChannel("input", inputBindingProperties); + + Binding producerBinding = binder.bindProducer("foo.0y", moduleOutputChannel, + outputBindingProperties.getProducer()); + Binding consumerBinding = binder.bindConsumer("foo.0y", "testSendAndReceiveJavaSerialization", moduleInputChannel, + createConsumerProperties()); + SerializableFoo foo = new SerializableFoo(); + Message message = MessageBuilder.withPayload(foo).setHeader(MessageHeaders.CONTENT_TYPE, MessageConverterUtils.X_JAVA_SERIALIZED_OBJECT) + .build(); + // Let the consumer actually bind to the producer before sending a msg + binderBindUnbindLatency(); + + CountDownLatch latch = new CountDownLatch(1); + AtomicReference> inboundMessageRef = new AtomicReference>(); + moduleInputChannel.subscribe(new MessageHandler() { + + @Override + public void handleMessage(Message message) throws MessagingException { + try { + inboundMessageRef.set((Message) message); + } + finally { + latch.countDown(); + } + } + }); + + moduleOutputChannel.send(message); + Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message"); + + assertThat(inboundMessageRef.get().getPayload()).isInstanceOf(SerializableFoo.class); + assertThat(inboundMessageRef.get().getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull(); + assertThat(inboundMessageRef.get().getHeaders().get(MessageHeaders.CONTENT_TYPE)).isEqualTo(MessageConverterUtils.X_JAVA_SERIALIZED_OBJECT); producerBinding.unbind(); consumerBinding.unbind(); } @Test + @SuppressWarnings("rawtypes") public void testSendAndReceiveMultipleTopics() throws Exception { Binder binder = getBinder(); @@ -161,15 +306,15 @@ public abstract class AbstractBinderTests producerBinding1 = binder.bindProducer("foo.x", + Binding producerBinding1 = binder.bindProducer("foo.xy", moduleOutputChannel1, createProducerProperties()); - Binding producerBinding2 = binder.bindProducer("foo.y", + Binding producerBinding2 = binder.bindProducer("foo.yz", moduleOutputChannel2, createProducerProperties()); - Binding consumerBinding1 = binder.bindConsumer("foo.x", + Binding consumerBinding1 = binder.bindConsumer("foo.xy", "testSendAndReceiveMultipleTopics", moduleInputChannel, createConsumerProperties()); - Binding consumerBinding2 = binder.bindConsumer("foo.y", + Binding consumerBinding2 = binder.bindConsumer("foo.yz", "testSendAndReceiveMultipleTopics", moduleInputChannel, createConsumerProperties()); @@ -206,6 +351,7 @@ public abstract class AbstractBinderTests producerBinding = binder.bindProducer("bar.0", moduleOutputChannel, producerBindingProperties.getProducer()); Binding consumerBinding = binder.bindConsumer("bar.0", @@ -224,10 +371,26 @@ public abstract class AbstractBinderTests message = MessageBuilder.withPayload("foo") .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN).build(); moduleOutputChannel.send(message); - Message inbound = receive(moduleInputChannel); - assertThat(inbound).isNotNull(); - assertThat(inbound.getPayload()).isEqualTo("foo".getBytes()); - assertThat(inbound.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString()) + CountDownLatch latch = new CountDownLatch(1); + AtomicReference> inboundMessageRef = new AtomicReference>(); + moduleInputChannel.subscribe(new MessageHandler() { + + @Override + public void handleMessage(Message message) throws MessagingException { + try { + inboundMessageRef.set((Message) message); + } + finally { + latch.countDown(); + } + } + }); + + moduleOutputChannel.send(message); + Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message"); + assertThat(inboundMessageRef.get()).isNotNull(); + assertThat(inboundMessageRef.get().getPayload()).isEqualTo("foo"); + assertThat(inboundMessageRef.get().getHeaders().get(MessageHeaders.CONTENT_TYPE).toString()) .isEqualTo(MimeTypeUtils.TEXT_PLAIN_VALUE); producerBinding.unbind(); consumerBinding.unbind(); @@ -254,6 +417,12 @@ public abstract class AbstractBinderTests producerBinding = binder.bindProducer("bad.0a", + moduleOutputChannel, createProducerProperties()); + Binding consumerBinding = binder.bindConsumer("bad.0a", "test-1", + moduleInputChannel, createConsumerProperties()); + + Station station = new Station(); + Message message = MessageBuilder.withPayload(station).build(); + moduleInputChannel.subscribe(handler); + moduleOutputChannel.send(message); + + QueueChannel replyChannel = (QueueChannel) handler.getOutputChannel(); + + Message replyMessage = replyChannel.receive(5000); + assertTrue(replyMessage.getPayload() instanceof Station); + producerBinding.unbind(); + consumerBinding.unbind(); + } + + @SuppressWarnings("rawtypes") + @Test + public void testSendPojoReceivePojoKryoWithStreamListener() throws Exception { + StreamListenerMessageHandler handler = this.buildStreamListener( + AbstractBinderTests.class, "echoStation", Station.class); + + Binder binder = getBinder(); + DirectChannel moduleOutputChannel = createBindableChannel("output", + new BindingProperties()); + DirectChannel moduleInputChannel = createBindableChannel("input", + new BindingProperties()); + Binding producerBinding = binder.bindProducer("bad.0b", + moduleOutputChannel, createProducerProperties()); + Binding consumerBinding = binder.bindConsumer("bad.0b", "test-2", + moduleInputChannel, createConsumerProperties()); + + 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); + assertTrue(replyMessage.getPayload() instanceof Station); + producerBinding.unbind(); + consumerBinding.unbind(); + } + + @SuppressWarnings("rawtypes") + @Test(expected = MessageHandlingException.class) + public void testStreamListenerJavaSerializationNonSerializable() throws Exception { + Binder binder = getBinder(); + DirectChannel moduleOutputChannel = createBindableChannel("output", + new BindingProperties()); + DirectChannel moduleInputChannel = createBindableChannel("input", + new BindingProperties()); + Binding producerBinding = binder.bindProducer("bad.0c", + moduleOutputChannel, createProducerProperties()); + Binding consumerBinding = binder.bindConsumer("bad.0c", "test-3", + moduleInputChannel, createConsumerProperties()); + 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 { + StreamListenerMessageHandler handler = this.buildStreamListener( + AbstractBinderTests.class, "echoStation", Station.class); + Binder binder = getBinder(); + DirectChannel moduleOutputChannel = createBindableChannel("output", + new BindingProperties()); + DirectChannel moduleInputChannel = createBindableChannel("input", + new BindingProperties()); + Binding producerBinding = binder.bindProducer("bad.0d", + moduleOutputChannel, createProducerProperties()); + Binding consumerBinding = binder.bindConsumer("bad.0d", "test-4", + moduleInputChannel, createConsumerProperties()); + + String value = "{\"readings\":[{\"stationid\":\"fgh\"," + + "\"customerid\":\"12345\",\"timestamp\":null},{\"stationid\":\"hjk\",\"customerid\":\"222\",\"timestamp\":null}]}"; + + Message message = MessageBuilder.withPayload(value) + .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON) + .build(); + moduleInputChannel.subscribe(handler); + moduleOutputChannel.send(message); + + QueueChannel channel = (QueueChannel) handler.getOutputChannel(); + + Message reply = (Message) channel.receive(5000); + + assertNotNull(reply); + assertTrue(reply.getPayload() instanceof Station); + producerBinding.unbind(); + consumerBinding.unbind(); + } + + @SuppressWarnings("rawtypes") + @Test + public void testSendJsonReceiveJsonWithStreamListener() throws Exception { + StreamListenerMessageHandler handler = this.buildStreamListener( + AbstractBinderTests.class, "echoStationString", String.class); + Binder binder = getBinder(); + DirectChannel moduleOutputChannel = createBindableChannel("output", + new BindingProperties()); + DirectChannel moduleInputChannel = createBindableChannel("input", + new BindingProperties()); + Binding producerBinding = binder.bindProducer("bad.0e", + moduleOutputChannel, createProducerProperties()); + Binding consumerBinding = binder.bindConsumer("bad.0e", "test-5", + moduleInputChannel, createConsumerProperties()); + + String value = "{\"readings\":[{\"stationid\":\"fgh\"," + + "\"customerid\":\"12345\",\"timestamp\":null},{\"stationid\":\"hjk\",\"customerid\":\"222\",\"timestamp\":null}]}"; + + Message message = MessageBuilder.withPayload(value) + .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON) + .build(); + moduleInputChannel.subscribe(handler); + moduleOutputChannel.send(message); + + QueueChannel channel = (QueueChannel) handler.getOutputChannel(); + + Message reply = (Message) channel.receive(5000); + + assertNotNull(reply); + assertTrue(reply.getPayload() instanceof String); + producerBinding.unbind(); + consumerBinding.unbind(); + } + + @SuppressWarnings("rawtypes") + @Test + public void testSendPojoReceivePojoWithStreamListener() throws Exception { + StreamListenerMessageHandler handler = this.buildStreamListener( + AbstractBinderTests.class, "echoStation", Station.class); + Binder binder = getBinder(); + DirectChannel moduleOutputChannel = createBindableChannel("output", + new BindingProperties()); + DirectChannel moduleInputChannel = createBindableChannel("input", + new BindingProperties()); + Binding producerBinding = binder.bindProducer("bad.0f", + moduleOutputChannel, createProducerProperties()); + Binding consumerBinding = binder.bindConsumer("bad.0f", "test-6", + moduleInputChannel, createConsumerProperties()); + + Readings r1 = new Readings(); + r1.setCustomerid("123"); + r1.setStationid("XYZ"); + Readings r2 = new Readings(); + r2.setCustomerid("546"); + r2.setStationid("ABC"); + Station station = new Station(); + station.setReadings(Arrays.asList(r1, r2)); + Message message = MessageBuilder.withPayload(station) + .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON) + .build(); + moduleInputChannel.subscribe(handler); + moduleOutputChannel.send(message); + + QueueChannel channel = (QueueChannel) handler.getOutputChannel(); + + Message reply = (Message) channel.receive(5000); + + assertNotNull(reply); + assertTrue(reply.getPayload() instanceof Station); + producerBinding.unbind(); + consumerBinding.unbind(); + } + + private static boolean equalTypeAndSubType(MimeType m1, MimeType m2) { + return m1 != null && m2 != null && m1.getType().equalsIgnoreCase(m2.getType()) + && m1.getSubtype().equalsIgnoreCase(m2.getSubtype()); + } + + @SuppressWarnings("unused") // it is used via reflection + private Station echoStation(Station station) { + return station; + } + + @SuppressWarnings("unused") // it is used via reflection + private String echoStationString(String station) { + return station; + } + + private StreamListenerMessageHandler buildStreamListener(Class handlerClass, + String handlerMethodName, Class... parameters) throws Exception { + String channelName = "reply_" + System.nanoTime(); + GenericApplicationContext context = new GenericApplicationContext(); + context.getBeanFactory().registerSingleton(channelName, new QueueChannel()); + + Method m = ReflectionUtils.findMethod(handlerClass, handlerMethodName, + parameters); + InvocableHandlerMethod method = new InvocableHandlerMethod(this, m); + HandlerMethodArgumentResolverComposite resolver = new HandlerMethodArgumentResolverComposite(); + CompositeMessageConverterFactory factory = new CompositeMessageConverterFactory(); + resolver.addResolver(new PayloadArgumentResolver( + factory.getMessageConverterForAllRegistered())); + method.setMessageMethodArgumentResolvers(resolver); + Constructor c = ReflectionUtils.accessibleConstructor( + StreamListenerMessageHandler.class, InvocableHandlerMethod.class, + boolean.class, String[].class); + StreamListenerMessageHandler handler = (StreamListenerMessageHandler) c + .newInstance(method, false, new String[] {}); + handler.setOutputChannelName(channelName); + handler.setBeanFactory(context); + handler.afterPropertiesSet(); + context.refresh(); + return handler; + } + + public static class Station { + + List readings = new ArrayList<>(); + + public List getReadings() { + return readings; + } + + public void setReadings(List readings) { + this.readings = readings; + } + + @SuppressWarnings("serial") + public static class Readings implements Serializable { + public String stationid; + public String customerid; + public String timestamp; + + public String getStationid() { + return stationid; + } + + public void setStationid(String stationid) { + this.stationid = stationid; + } + + public String getCustomerid() { + return customerid; + } + + public void setCustomerid(String customerid) { + this.customerid = customerid; + } + + public String getTimestamp() { + return timestamp; + } + + public void setTimestamp(String timestamp) { + this.timestamp = timestamp; + } + } + } } diff --git a/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractTestBinder.java b/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractTestBinder.java index 834ea5fb5..1cab5e15e 100644 --- a/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractTestBinder.java +++ b/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractTestBinder.java @@ -50,7 +50,6 @@ public abstract class AbstractTestBinder bindProducer(String name, MessageChannel moduleOutputChannel, PP properties) { - this.checkChannelIsConfigured(moduleOutputChannel); queues.add(name); return binder.bindProducer(name, moduleOutputChannel, properties); } diff --git a/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/SerializableFoo.java b/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/SerializableFoo.java new file mode 100644 index 000000000..5e6649977 --- /dev/null +++ b/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/SerializableFoo.java @@ -0,0 +1,34 @@ +/* + * Copyright 2017 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder; + + +import java.io.Serializable; + +/** + * + * @author Oleg Zhurakousky + * + */ +public class SerializableFoo implements Serializable{ + + /** + * + */ + private static final long serialVersionUID = 1L; + +} diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/ContentTypeOutboundSourceTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/ContentTypeOutboundSourceTests.java index 1f9e6b683..2be956abd 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/ContentTypeOutboundSourceTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/ContentTypeOutboundSourceTests.java @@ -22,7 +22,6 @@ import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.cloud.stream.annotation.Bindings; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.binder.BinderFactory; import org.springframework.cloud.stream.messaging.Source; @@ -39,13 +38,13 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Ilayaperumal Gopinathan * @author Vinicius Carvalho + * @author Oleg Zhurakousky */ @RunWith(SpringJUnit4ClassRunner.class) @SpringBootTest(classes = { ContentTypeOutboundSourceTests.TestSource.class }) public class ContentTypeOutboundSourceTests { @Autowired - @Bindings(TestSource.class) private Source testSource; @Autowired @@ -55,14 +54,11 @@ public class ContentTypeOutboundSourceTests { @SuppressWarnings("unchecked") public void testMessageHeaderWhenNoExplicitContentTypeOnMessage() throws Exception { testSource.output().send(MessageBuilder.withPayload("{\"message\":\"Hi\"}").setHeader(MessageHeaders.CONTENT_TYPE,"text/plain").build()); - Message received = ((TestSupportBinder) binderFactory.getBinder(null, + Message received = (Message) ((TestSupportBinder) binderFactory.getBinder(null, MessageChannel.class)) .messageCollector().forChannel(testSource.output()).poll(); assertThat(received.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString()).contains("text/plain"); - Object payload = received.getPayload(); - assertThat(payload.getClass().isAssignableFrom(byte[].class)).isTrue(); - byte[] contents = (byte[])payload; - assertThat("{\"message\":\"Hi\"}").isEqualTo(new String(contents)); + assertThat("{\"message\":\"Hi\"}").isEqualTo(received.getPayload()); } @EnableBinding(Source.class) diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/CustomHeaderPropagationTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/CustomHeaderPropagationTests.java index 050e5b83a..77d91ff20 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/CustomHeaderPropagationTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/CustomHeaderPropagationTests.java @@ -39,6 +39,7 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Marius Bogoevici + * @author Oleg Zhurakousky */ @RunWith(SpringJUnit4ClassRunner.class) @SpringBootTest(classes = CustomHeaderPropagationTests.HeaderPropagationProcessor.class, @@ -65,13 +66,13 @@ public class CustomHeaderPropagationTests { .setHeader("bar", "barValue") .build()); @SuppressWarnings("unchecked") - Message received = ((TestSupportBinder) binderFactory.getBinder(null, MessageChannel.class)) + Message received = (Message) ((TestSupportBinder) binderFactory.getBinder(null, MessageChannel.class)) .messageCollector().forChannel(testProcessor.output()).poll(10, TimeUnit.SECONDS); assertThat(received).isNotNull(); assertThat(received.getHeaders()).containsEntry("foo", "fooValue"); assertThat(received.getHeaders()).doesNotContainKey("bar"); assertThat(received.getHeaders()).containsKeys(MessageHeaders.CONTENT_TYPE); - assertThat(new String((byte[])received.getPayload())).isEqualTo("{'name':'foo'}"); + assertThat(new String(received.getPayload())).isEqualTo("{'name':'foo'}"); } @EnableBinding(Processor.class) diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/DefaultHeaderPropagationTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/DefaultHeaderPropagationTests.java index 049b81ebb..32657c4ff 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/DefaultHeaderPropagationTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/DefaultHeaderPropagationTests.java @@ -39,6 +39,7 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Marius Bogoevici + * @author Oleg Zhurakousky */ @RunWith(SpringJUnit4ClassRunner.class) @SpringBootTest(classes = DefaultHeaderPropagationTests.HeaderPropagationProcessor.class, @@ -59,13 +60,13 @@ public class DefaultHeaderPropagationTests { .setHeader("bar", "barValue") .build()); @SuppressWarnings("unchecked") - Message received = ((TestSupportBinder) binderFactory.getBinder(null, MessageChannel.class)) + Message received = (Message) ((TestSupportBinder) binderFactory.getBinder(null, MessageChannel.class)) .messageCollector().forChannel(testProcessor.output()).poll(1, TimeUnit.SECONDS); assertThat(received).isNotNull(); assertThat(received.getHeaders()).containsEntry("foo", "fooValue"); assertThat(received.getHeaders()).containsEntry("bar", "barValue"); assertThat(received.getHeaders()).containsKeys(MessageHeaders.CONTENT_TYPE); - assertThat(received.getPayload()).isEqualTo("{'name':'foo'}".getBytes()); + assertThat(received.getPayload()).isEqualTo("{'name':'foo'}"); } @EnableBinding(Processor.class) diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/DefaultHeaderPropagationWithApplicationProvidedHeaderTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/DefaultHeaderPropagationWithApplicationProvidedHeaderTests.java index 43b5fe925..a0cee605e 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/DefaultHeaderPropagationWithApplicationProvidedHeaderTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/DefaultHeaderPropagationWithApplicationProvidedHeaderTests.java @@ -33,12 +33,14 @@ import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHeaders; -import org.springframework.messaging.converter.MessageConversionException; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import static org.junit.Assert.assertEquals; + /** * @author Marius Bogoevici + * @author Oleg Zhurakousky */ @RunWith(SpringJUnit4ClassRunner.class) @SpringBootTest(classes = DefaultHeaderPropagationWithApplicationProvidedHeaderTests.HeaderPropagationProcessor.class, @@ -51,17 +53,18 @@ public class DefaultHeaderPropagationWithApplicationProvidedHeaderTests { @Autowired private BinderFactory binderFactory; - @Test(expected = MessageConversionException.class) - public void testFailedonCustomContentTypeWithoutConverter() throws Exception { + @Test + public void testHeaderPropagationIfSetByApplication() throws Exception { testProcessor.input().send(MessageBuilder.withPayload("{'name':'foo'}") .setHeader(MessageHeaders.CONTENT_TYPE, "application/json") .setHeader("foo", "fooValue") .setHeader("bar", "barValue") .build()); @SuppressWarnings("unchecked") - Message received = ((TestSupportBinder) binderFactory.getBinder(null, MessageChannel.class)) + Message received = (Message) ((TestSupportBinder) binderFactory.getBinder(null, MessageChannel.class)) .messageCollector().forChannel(testProcessor.output()).poll(1, TimeUnit.SECONDS); - + assertEquals("fooValue", received.getHeaders().get("foo")); + assertEquals("barValue", received.getHeaders().get("bar")); } @EnableBinding(Processor.class) diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/DeserializeJSONToJavaTypeTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/DeserializeJSONToJavaTypeTests.java index 5d7d8123a..4ca9e380a 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/DeserializeJSONToJavaTypeTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/DeserializeJSONToJavaTypeTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -16,7 +16,6 @@ package org.springframework.cloud.stream.config; -import java.util.List; import java.util.concurrent.TimeUnit; import org.junit.Test; @@ -35,13 +34,13 @@ import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.converter.MessageConverter; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import static org.assertj.core.api.Assertions.assertThat; /** * @author Marius Bogoevici + * @author Oleg Zhurakousky */ @RunWith(SpringJUnit4ClassRunner.class) @SpringBootTest(classes = DeserializeJSONToJavaTypeTests.FooProcessor.class) @@ -53,19 +52,15 @@ public class DeserializeJSONToJavaTypeTests { @Autowired private BinderFactory binderFactory; - @Autowired - private List customMessageConverters; - @Test public void testMessageDeserialized() throws Exception { testProcessor.input().send( MessageBuilder.withPayload("{\"name\":\"Bar\"}").setHeader("contentType", "application/json").build()); @SuppressWarnings("unchecked") - Message received = ((TestSupportBinder) binderFactory.getBinder(null, MessageChannel.class)) + Message received = (Message) ((TestSupportBinder) binderFactory.getBinder(null, MessageChannel.class)) .messageCollector().forChannel(testProcessor.output()).poll(1, TimeUnit.SECONDS); assertThat(received).isNotNull(); - assertThat(received.getPayload()).isInstanceOf(byte[].class); - assertThat((byte[]) received.getPayload()).isEqualTo("{\"name\":\"Bar\"}".getBytes()); + assertThat(received.getPayload()).isEqualTo("{\"name\":\"Bar\"}"); } @EnableBinding(Processor.class) diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/InboundJsonToTupleConversionTest.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/InboundJsonToTupleConversionTest.java index d5e9d3d33..51014372f 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/InboundJsonToTupleConversionTest.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/InboundJsonToTupleConversionTest.java @@ -41,6 +41,7 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Marius Bogoevici + * @author Oleg Zhurakousky */ @RunWith(SpringJUnit4ClassRunner.class) @SpringBootTest(classes = InboundJsonToTupleConversionTest.FooProcessor.class) @@ -57,11 +58,11 @@ public class InboundJsonToTupleConversionTest { testProcessor.input().send(MessageBuilder.withPayload("{'name':'foo'}") .build()); @SuppressWarnings("unchecked") - Message received = ((TestSupportBinder) binderFactory.getBinder(null, MessageChannel.class)) + Message received = (Message) ((TestSupportBinder) binderFactory.getBinder(null, MessageChannel.class)) .messageCollector().forChannel(testProcessor.output()).poll(1, TimeUnit.SECONDS); assertThat(received).isNotNull(); - assertThat(TupleBuilder.fromString(new String((byte[])received.getPayload()))).isEqualTo(TupleBuilder.tuple().of("name", "foo")); + assertThat(TupleBuilder.fromString(new String(received.getPayload()))).isEqualTo(TupleBuilder.tuple().of("name", "foo")); } @EnableBinding(Processor.class) diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/LegacyContentTypeTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/LegacyContentTypeTests.java index 37e87a52d..5c6adfb64 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/LegacyContentTypeTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/LegacyContentTypeTests.java @@ -39,6 +39,7 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Soby Chacko + * @author Oleg Zhurakousky */ @RunWith(SpringJUnit4ClassRunner.class) @SpringBootTest(classes = { LegacyContentTypeTests.LegacyTestSink.class}) @@ -53,9 +54,9 @@ public class LegacyContentTypeTests { MessageHandler messageHandler = new MessageHandler() { @Override public void handleMessage(Message message) throws MessagingException { - assertThat(message.getPayload()).isInstanceOf(byte[].class); - assertThat(message.getPayload()).isEqualTo("{\"message\":\"Hi\"}".getBytes()); - assertThat(message.getHeaders().get(MessageHeaders.CONTENT_TYPE)).isEqualTo("application/json"); + assertThat(message.getPayload()).isInstanceOf(String.class); + assertThat(message.getPayload()).isEqualTo("{\"message\":\"Hi\"}"); + assertThat(message.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString()).isEqualTo("application/json"); latch.countDown(); } }; diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerAnnotatedMethodArgumentsTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerAnnotatedMethodArgumentsTests.java index cc4ec8cf9..28a203ce4 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerAnnotatedMethodArgumentsTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerAnnotatedMethodArgumentsTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -37,6 +37,7 @@ import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.handler.annotation.Header; import org.springframework.messaging.handler.annotation.Headers; import org.springframework.messaging.handler.annotation.Payload; +import org.springframework.util.MimeType; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.Assert.fail; @@ -45,6 +46,7 @@ import static org.springframework.cloud.stream.binding.StreamListenerErrorMessag /** * @author Marius Bogoevici * @author Ilayaperumal Gopinathan + * @author Oleg Zhurakousky */ public class StreamListenerAnnotatedMethodArgumentsTests { @@ -66,8 +68,8 @@ public class StreamListenerAnnotatedMethodArgumentsTests { assertThat(testPojoWithAnnotatedArguments.receivedArguments.get(0)).hasFieldOrPropertyWithValue("foo", "barbar" + id); assertThat(testPojoWithAnnotatedArguments.receivedArguments.get(1)).isInstanceOf(Map.class); - assertThat((Map) testPojoWithAnnotatedArguments.receivedArguments.get(1)) - .containsEntry(MessageHeaders.CONTENT_TYPE, "application/json"); + assertThat((Map) testPojoWithAnnotatedArguments.receivedArguments.get(1)) + .containsEntry(MessageHeaders.CONTENT_TYPE, MimeType.valueOf("application/json")); assertThat((Map) testPojoWithAnnotatedArguments.receivedArguments.get(1)) .containsEntry("testHeader", "testValue"); assertThat(testPojoWithAnnotatedArguments.receivedArguments.get(2)).isEqualTo("application/json"); diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerHandlerBeanTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerHandlerBeanTests.java index ba9c823eb..1db407b56 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerHandlerBeanTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerHandlerBeanTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -50,6 +50,7 @@ import static org.assertj.core.api.Assertions.assertThat; * @author Marius Bogoevici * @author Ilayaperumal Gopinathan * @author Vinicius Carvalho + * @author Oleg Zhurakousky */ @RunWith(Parameterized.class) public class StreamListenerHandlerBeanTests { @@ -61,7 +62,7 @@ public class StreamListenerHandlerBeanTests { } @Parameterized.Parameters - public static Collection InputConfigs() { + public static Collection InputConfigs() { return Arrays.asList(TestHandlerBeanWithSendTo.class, TestHandlerBean2.class); } @@ -81,10 +82,10 @@ public class StreamListenerHandlerBeanTests { Assertions.assertThat(handlerBean.receivedPojos).hasSize(1); Assertions.assertThat(handlerBean.receivedPojos.get(0)).hasFieldOrPropertyWithValue("foo", "barbar" + id); - Message message = (Message) collector.forChannel( + Message message = (Message) collector.forChannel( processor.output()).poll(1, TimeUnit.SECONDS); assertThat(message).isNotNull(); - assertThat(new String(message.getPayload())).isEqualTo("{\"bar\":\"barbar" + id + "\"}"); + assertThat(message.getPayload()).isEqualTo("{\"bar\":\"barbar" + id + "\"}"); assertThat(message.getHeaders().get(MessageHeaders.CONTENT_TYPE, MimeType.class) .includes(MimeTypeUtils.APPLICATION_JSON)); context.close(); diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerHandlerMethodTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerHandlerMethodTests.java index c3ba673d9..811bb0353 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerHandlerMethodTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerHandlerMethodTests.java @@ -16,7 +16,6 @@ package org.springframework.cloud.stream.config; -import java.util.UUID; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -65,6 +64,7 @@ import static org.springframework.cloud.stream.binding.StreamListenerErrorMessag * @author Ilayaperumal Gopinathan * @author Gary Russell * @author Vinicius Carvalho + * @author Oleg Zhurakousky */ public class StreamListenerHandlerMethodTests { @@ -80,6 +80,7 @@ public class StreamListenerHandlerMethodTests { } } + @SuppressWarnings("unchecked") @Test public void testMethodWithObjectAsMethodArgument() throws Exception { ConfigurableApplicationContext context = SpringApplication.run(TestMethodWithObjectAsMethodArgument.class, @@ -91,12 +92,13 @@ public class StreamListenerHandlerMethodTests { final String testMessage = "testing"; processor.input().send(MessageBuilder.withPayload(testMessage).build()); MessageCollector messageCollector = context.getBean(MessageCollector.class); - Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); + Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); assertThat(result).isNotNull(); - assertThat(new String(result.getPayload())).isEqualTo(testMessage.toUpperCase()); + assertThat(result.getPayload()).isEqualTo(testMessage.toUpperCase()); context.close(); } + @SuppressWarnings("unchecked") @Test /** * @since 2.0 : This test is an example of the new behavior of 2.0 when it comes to contentType handling. @@ -115,13 +117,14 @@ public class StreamListenerHandlerMethodTests { .setHeader("foo", "bar") .build()); MessageCollector messageCollector = context.getBean(MessageCollector.class); - Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); + Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); assertThat(result).isNotNull(); - assertThat(new String(result.getPayload())).isEqualTo(testMessage.toUpperCase()); + assertThat(result.getPayload()).isEqualTo(testMessage.toUpperCase()); assertThat(result.getHeaders().get("foo")).isEqualTo("bar"); context.close(); } + @SuppressWarnings("unchecked") @Test public void testMethodHeadersNotPropagatged() throws Exception { ConfigurableApplicationContext context = SpringApplication.run(TestMethodHeadersNotPropagated.class, @@ -135,15 +138,16 @@ public class StreamListenerHandlerMethodTests { .setHeader("foo", "bar") .build()); MessageCollector messageCollector = context.getBean(MessageCollector.class); - Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); + Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); assertThat(result).isNotNull(); - assertThat(new String(result.getPayload())).isEqualTo(testMessage.toUpperCase()); + assertThat(result.getPayload()).isEqualTo(testMessage.toUpperCase()); assertThat(result.getHeaders().get("foo")).isNull(); context.close(); } //TODO: Handle dynamic destinations and contentType + @SuppressWarnings("unchecked") public void testStreamListenerMethodWithTargetBeanFromOutside() throws Exception { ConfigurableApplicationContext context = SpringApplication .run(TestStreamListenerMethodWithTargetBeanFromOutside.class, "--server.port=0", @@ -156,10 +160,10 @@ public class StreamListenerHandlerMethodTests { DirectChannel directChannel = (DirectChannel) context.getBean(testMessageToSend.toUpperCase(), MessageChannel.class); MessageCollector messageCollector = context.getBean(MessageCollector.class); - Message result = (Message) messageCollector.forChannel(directChannel).poll(1000, TimeUnit.MILLISECONDS); + Message result = (Message) messageCollector.forChannel(directChannel).poll(1000, TimeUnit.MILLISECONDS); sink.input().send(MessageBuilder.withPayload(testMessageToSend).build()); assertThat(result).isNotNull(); - assertThat(new String(result.getPayload())).isEqualTo(testMessageToSend.toUpperCase()); + assertThat(result.getPayload()).isEqualTo(testMessageToSend.toUpperCase()); context.close(); } @@ -306,12 +310,11 @@ public class StreamListenerHandlerMethodTests { Processor processor = context.getBean(Processor.class); StreamListenerTestUtils.FooInboundChannel1 inboundChannel2 = context .getBean(StreamListenerTestUtils.FooInboundChannel1.class); - String id = UUID.randomUUID().toString(); final CountDownLatch latch = new CountDownLatch(2); ((SubscribableChannel) processor.output()).subscribe(new MessageHandler() { @Override public void handleMessage(Message message) throws MessagingException { - Assert.isTrue(message.getPayload().equals("footesting") || message.getPayload().equals("BARTESTING")); + Assert.isTrue(message.getPayload().equals("footesting") || message.getPayload().equals("BARTESTING"), "Assert failed"); latch.countDown(); } }); @@ -329,23 +332,22 @@ public class StreamListenerHandlerMethodTests { "--server.port=0", "--spring.jmx.enabled=false"); Processor processor = context.getBean(Processor.class); - String id = UUID.randomUUID().toString(); StreamListenerTestUtils.FooOutboundChannel1 source2 = context .getBean(StreamListenerTestUtils.FooOutboundChannel1.class); final CountDownLatch latch = new CountDownLatch(2); ((SubscribableChannel) processor.output()).subscribe(new MessageHandler() { @Override public void handleMessage(Message message) throws MessagingException { - Assert.isTrue(message.getPayload().equals("testing")); - Assert.isTrue(message.getHeaders().get("output").equals("output2")); + Assert.isTrue(message.getPayload().equals("testing"), "Assert failed"); + Assert.isTrue(message.getHeaders().get("output").equals("output2"), "Assert failed"); latch.countDown(); } }); ((SubscribableChannel) source2.output()).subscribe(new MessageHandler() { @Override public void handleMessage(Message message) throws MessagingException { - Assert.isTrue(message.getPayload().equals("TESTING")); - Assert.isTrue(message.getHeaders().get("output").equals("output1")); + Assert.isTrue(message.getPayload().equals("TESTING"), "Assert failed"); + Assert.isTrue(message.getHeaders().get("output").equals("output1"), "Assert failed"); latch.countDown(); } }); diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerMessageArgumentTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerMessageArgumentTests.java index a43756cd2..6c1a2b9ff 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerMessageArgumentTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerMessageArgumentTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -45,6 +45,7 @@ import static org.assertj.core.api.Assertions.assertThat; * @author Marius Bogoevici * @author Ilayaperumal Gopinathan * @author Vinicius Carvalho + * @author Oleg Zhurakousky */ @RunWith(Parameterized.class) public class StreamListenerMessageArgumentTests { @@ -56,7 +57,7 @@ public class StreamListenerMessageArgumentTests { } @Parameterized.Parameters - public static Collection InputConfigs() { + public static Collection InputConfigs() { return Arrays.asList(new Class[] { TestPojoWithMessageArgument1.class, TestPojoWithMessageArgument2.class }); } @@ -74,10 +75,10 @@ public class StreamListenerMessageArgumentTests { .getBean(TestPojoWithMessageArgument.class); assertThat(testPojoWithMessageArgument.receivedMessages).hasSize(1); assertThat(testPojoWithMessageArgument.receivedMessages.get(0).getPayload()).isEqualTo("barbar" + id); - Message message = (Message) collector + Message message = (Message) collector .forChannel(processor.output()).poll(1, TimeUnit.SECONDS); assertThat(message).isNotNull(); - assertThat(new String(message.getPayload())).contains("barbar" + id); + assertThat(message.getPayload()).contains("barbar" + id); context.close(); } diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerMethodReturnWithConversionTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerMethodReturnWithConversionTests.java index d345dad6f..b40dfe943 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerMethodReturnWithConversionTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerMethodReturnWithConversionTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -53,6 +53,7 @@ import static org.assertj.core.api.Assertions.assertThat; * @author Marius Bogoevici * @author Ilayaperumal Gopinathan * @author Vinicius Carvalho + * @author Oleg Zhurakousky * */ @RunWith(StreamListenerMethodReturnWithConversionTests.class) @@ -75,7 +76,7 @@ public class StreamListenerMethodReturnWithConversionTests extends Suite { } @Parameterized.Parameters - public static Collection InputConfigs() { + public static Collection InputConfigs() { return Arrays.asList(new Class[] { TestPojoWithMimeType1.class, TestPojoWithMimeType2.class }); } @@ -92,7 +93,7 @@ public class StreamListenerMethodReturnWithConversionTests extends Suite { TestPojoWithMimeType testPojoWithMimeType = context.getBean(TestPojoWithMimeType.class); Assertions.assertThat(testPojoWithMimeType.receivedPojos).hasSize(1); Assertions.assertThat(testPojoWithMimeType.receivedPojos.get(0)).hasFieldOrPropertyWithValue("foo", "barbar" + id); - Message message = (Message) collector.forChannel(processor.output()).poll(1, + Message message = (Message) collector.forChannel(processor.output()).poll(1, TimeUnit.SECONDS); assertThat(message).isNotNull(); assertThat(new String(message.getPayload())).isEqualTo("{\"bar\":\"barbar" + id + "\"}"); @@ -114,7 +115,7 @@ public class StreamListenerMethodReturnWithConversionTests extends Suite { } @Parameterized.Parameters - public static Collection InputConfigs() { + public static Collection InputConfigs() { return Arrays.asList(new Class[] { TestPojoWithMimeType1.class, TestPojoWithMimeType2.class }); } @@ -130,7 +131,7 @@ public class StreamListenerMethodReturnWithConversionTests extends Suite { TestPojoWithMimeType testPojoWithMimeType = context.getBean(TestPojoWithMimeType.class); Assertions.assertThat(testPojoWithMimeType.receivedPojos).hasSize(1); Assertions.assertThat(testPojoWithMimeType.receivedPojos.get(0)).hasFieldOrPropertyWithValue("foo", "barbar" + id); - Message message = (Message) collector + Message message = (Message) collector .forChannel(processor.output()).poll(1, TimeUnit.SECONDS); assertThat(message).isNotNull(); diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerMethodWithReturnMessageTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerMethodWithReturnMessageTests.java index 3b24535c1..74e33c764 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerMethodWithReturnMessageTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerMethodWithReturnMessageTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -46,6 +46,7 @@ import static org.assertj.core.api.Assertions.assertThat; * @author Marius Bogoevici * @author Ilayaperumal Gopinathan * @author Vinicius Carvalho + * @author Oleg Zhurakousky */ @RunWith(Parameterized.class) public class StreamListenerMethodWithReturnMessageTests { @@ -57,7 +58,7 @@ public class StreamListenerMethodWithReturnMessageTests { } @Parameterized.Parameters - public static Collection InputConfigs() { + public static Collection InputConfigs() { return Arrays.asList(new Class[] { TestPojoWithMessageReturn1.class, TestPojoWithMessageReturn2.class }); } @@ -76,10 +77,10 @@ public class StreamListenerMethodWithReturnMessageTests { .getBean(TestPojoWithMessageReturn.class); Assertions.assertThat(testPojoWithMessageReturn.receivedPojos).hasSize(1); Assertions.assertThat(testPojoWithMessageReturn.receivedPojos.get(0)).hasFieldOrPropertyWithValue("foo", "barbar" + id); - Message message = (Message) collector + Message message = (Message) collector .forChannel(processor.output()).poll(1, TimeUnit.SECONDS); assertThat(message).isNotNull(); - assertThat(new String(message.getPayload())).contains("barbar" + id); + assertThat(message.getPayload()).contains("barbar" + id); context.close(); } diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerMethodWithReturnValueTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerMethodWithReturnValueTests.java index 476d3480f..0c622f363 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerMethodWithReturnValueTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerMethodWithReturnValueTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -45,6 +45,7 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Marius Bogoevici * @author Ilayaperumal Gopinathan + * @author Oleg Zhurakousky */ @RunWith(Parameterized.class) public class StreamListenerMethodWithReturnValueTests { @@ -56,7 +57,7 @@ public class StreamListenerMethodWithReturnValueTests { } @Parameterized.Parameters - public static Collection InputConfigs() { + public static Collection InputConfigs() { return Arrays.asList(new Class[] { TestStringProcessor1.class, TestStringProcessor2.class }); } @@ -71,14 +72,14 @@ public class StreamListenerMethodWithReturnValueTests { processor.input() .send(MessageBuilder.withPayload("{\"foo\":\"barbar" + id + "\"}") .setHeader("contentType", "application/json").build()); - Message message = (Message) collector + Message message = (Message) collector .forChannel(processor.output()).poll(1, TimeUnit.SECONDS); TestStringProcessor testStringProcessor = context .getBean(TestStringProcessor.class); Assertions.assertThat(testStringProcessor.receivedPojos).hasSize(1); Assertions.assertThat(testStringProcessor.receivedPojos.get(0)).hasFieldOrPropertyWithValue("foo", "barbar" + id); assertThat(message).isNotNull(); - assertThat(new String(message.getPayload())).contains("barbar" + id); + assertThat(message.getPayload()).contains("barbar" + id); context.close(); } diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerWithAnnotatedInputOutputArgsTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerWithAnnotatedInputOutputArgsTests.java index d2f5626be..f704ca866 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerWithAnnotatedInputOutputArgsTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerWithAnnotatedInputOutputArgsTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -46,6 +46,7 @@ import static org.springframework.cloud.stream.binding.StreamListenerErrorMessag * @author Marius Bogoevici * @author Ilayaperumal Gopinathan * @author Vinicius Carvalho + * @author Oleg Zhurakousky */ public class StreamListenerWithAnnotatedInputOutputArgsTests { @@ -86,14 +87,14 @@ public class StreamListenerWithAnnotatedInputOutputArgsTests { sendMessageAndValidate(context); } + @SuppressWarnings("unchecked") private void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException { - @SuppressWarnings("unchecked") Processor processor = context.getBean(Processor.class); processor.input().send(MessageBuilder.withPayload("hello").setHeader("contentType", "text/plain").build()); MessageCollector messageCollector = context.getBean(MessageCollector.class); - Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); + Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); assertThat(result).isNotNull(); - assertThat(new String(result.getPayload())).isEqualTo("HELLO"); + assertThat(result.getPayload()).isEqualTo("HELLO"); context.close(); } diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerWithConditionsTest.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerWithConditionsTest.java index 73cc0a261..a77991e8a 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerWithConditionsTest.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerWithConditionsTest.java @@ -44,7 +44,6 @@ import static org.assertj.core.api.Assertions.fail; public class StreamListenerWithConditionsTest { @Test - @SuppressWarnings("unchecked") public void testAnnotatedArgumentsWithConditionalClass() throws Exception { ConfigurableApplicationContext context = SpringApplication.run(TestPojoWithAnnotatedArguments.class, "--server.port=0"); @@ -72,7 +71,6 @@ public class StreamListenerWithConditionsTest { } @Test - @SuppressWarnings("unchecked") public void testConditionalFailsWithReturnValue() throws Exception { try { ConfigurableApplicationContext context = SpringApplication.run( @@ -89,7 +87,6 @@ public class StreamListenerWithConditionsTest { } @Test - @SuppressWarnings("unchecked") public void testConditionalFailsWithDeclarativeMethod() throws Exception { try { ConfigurableApplicationContext context = SpringApplication.run( diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/TextPlainConversionTest.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/TextPlainConversionTest.java index 0f8abc8fe..c2411e9d1 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/TextPlainConversionTest.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/TextPlainConversionTest.java @@ -39,6 +39,7 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Marius Bogoevici + * @author Oleg Zhurakousky * * @since 1.2 */ @@ -56,30 +57,30 @@ public class TextPlainConversionTest { public void testTextPlainConversionOnOutput() throws Exception { testProcessor.input().send(MessageBuilder.withPayload("Bar").build()); @SuppressWarnings("unchecked") - Message received = (Message) ((TestSupportBinder) binderFactory.getBinder(null, MessageChannel.class)) + Message received = (Message) ((TestSupportBinder) binderFactory.getBinder(null, MessageChannel.class)) .messageCollector().forChannel(testProcessor.output()).poll(1, TimeUnit.SECONDS); assertThat(received).isNotNull(); - assertThat(new String(received.getPayload())).isEqualTo("Foo{name='Bar'}"); + assertThat(received.getPayload()).isEqualTo("Foo{name='Bar'}"); } @Test public void testByteArrayConversionOnOutput() throws Exception { testProcessor.output().send(MessageBuilder.withPayload("Bar".getBytes()).build()); @SuppressWarnings("unchecked") - Message received = (Message) ((TestSupportBinder) binderFactory.getBinder(null, MessageChannel.class)) + Message received = (Message)((TestSupportBinder) binderFactory.getBinder(null, MessageChannel.class)) .messageCollector().forChannel(testProcessor.output()).poll(1, TimeUnit.SECONDS); assertThat(received).isNotNull(); - assertThat(new String(received.getPayload())).isEqualTo("Bar"); + assertThat(received.getPayload()).isEqualTo("Bar"); } @Test public void testTextPlainConversionOnInputAndOutput() throws Exception { testProcessor.input().send(MessageBuilder.withPayload(new Foo("Bar")).build()); @SuppressWarnings("unchecked") - Message received = (Message) ((TestSupportBinder) binderFactory.getBinder(null, MessageChannel.class)) + Message received = (Message) ((TestSupportBinder) binderFactory.getBinder(null, MessageChannel.class)) .messageCollector().forChannel(testProcessor.output()).poll(1, TimeUnit.SECONDS); assertThat(received).isNotNull(); - assertThat(new String(received.getPayload())).isEqualTo("Foo{name='Foo{name='Bar'}'}"); + assertThat(received.getPayload()).isEqualTo("Foo{name='Foo{name='Bar'}'}"); } @EnableBinding(Processor.class) diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/TextPlainToJsonConversionTest.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/TextPlainToJsonConversionTest.java index 1df9ca403..9a261a95e 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/TextPlainToJsonConversionTest.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/TextPlainToJsonConversionTest.java @@ -43,6 +43,7 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Marius Bogoevici * @author Vinicius Carvalho + * @author Oleg Zhurakousky * @since 1.2 */ @RunWith(SpringJUnit4ClassRunner.class) @@ -57,10 +58,11 @@ public class TextPlainToJsonConversionTest { private ObjectMapper mapper = new ObjectMapper(); + @SuppressWarnings("unchecked") @Test public void testNoContentTypeToJsonConversionOnInput() throws Exception { testProcessor.input().send(MessageBuilder.withPayload("{\"name\":\"Bar\"}").build()); - Message received = (Message) ((TestSupportBinder) binderFactory.getBinder(null, MessageChannel.class)) + Message received = (Message) ((TestSupportBinder) binderFactory.getBinder(null, MessageChannel.class)) .messageCollector().forChannel(testProcessor.output()).poll(1, TimeUnit.SECONDS); assertThat(received).isNotNull(); Foo foo = mapper.readValue(received.getPayload(),Foo.class); @@ -75,10 +77,6 @@ public class TextPlainToJsonConversionTest { public void testTextPlainToJsonConversionOnInput() throws Exception { testProcessor.input().send(MessageBuilder.withPayload("{\"name\":\"Bar\"}") .setHeader(MessageHeaders.CONTENT_TYPE, "text/plain").build()); - Message received = ((TestSupportBinder) binderFactory.getBinder(null, MessageChannel.class)) - .messageCollector().forChannel(testProcessor.output()).poll(1, TimeUnit.SECONDS); - assertThat(received).isNotNull(); - assertThat(((Foo) received.getPayload()).getName()).isEqualTo("transformed-Bar"); } @EnableBinding(Processor.class) diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/aggregate/AggregateApplicationTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/aggregate/AggregateApplicationTests.java index d54d51dfd..d5deab0e0 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/aggregate/AggregateApplicationTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/aggregate/AggregateApplicationTests.java @@ -37,6 +37,7 @@ import static org.hamcrest.Matchers.notNullValue; /** * @author Ilayaperumal Gopinathan + * @author Oleg Zhurakousky */ @RunWith(SpringJUnit4ClassRunner.class) public class AggregateApplicationTests { @@ -49,10 +50,9 @@ public class AggregateApplicationTests { TestSupportBinder testSupportBinder = (TestSupportBinder) context.getBean(BinderFactory.class).getBinder(null, MessageChannel.class); MessageChannel processorOutput = testSupportBinder.getChannelForName("output"); - Message received = (Message) (testSupportBinder.messageCollector().forChannel(processorOutput) + Message received = (Message) (testSupportBinder.messageCollector().forChannel(processorOutput) .poll(5, TimeUnit.SECONDS)); Assert.assertThat(received, notNullValue()); - String payload = new String(received.getPayload()); - Assert.assertTrue(payload.endsWith("processed")); + Assert.assertTrue(received.getPayload().endsWith("processed")); } } diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/contentType/ContentTypeTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/contentType/ContentTypeTests.java index aa43e6561..f0e96783b 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/contentType/ContentTypeTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/contentType/ContentTypeTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2017 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. @@ -16,9 +16,7 @@ package org.springframework.cloud.stream.config.contentType; -import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; -import java.io.ObjectInputStream; import java.io.ObjectOutputStream; import java.io.Serializable; import java.util.LinkedList; @@ -30,6 +28,7 @@ import com.esotericsoftware.kryo.io.Output; import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonProperty; import com.fasterxml.jackson.databind.ObjectMapper; + import org.junit.Test; import org.springframework.boot.SpringApplication; @@ -44,6 +43,7 @@ import org.springframework.cloud.stream.test.binder.MessageCollector; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.handler.annotation.Headers; @@ -56,7 +56,9 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Vinicius Carvalho + * @author Oleg Zhurakousky */ +@SuppressWarnings("unchecked") public class ContentTypeTests { private ObjectMapper mapper = new ObjectMapper(); @@ -71,7 +73,7 @@ public class ContentTypeTests { Source source = context.getBean(Source.class); User user = new User("Alice"); source.output().send(MessageBuilder.withPayload(user).build()); - Message message = (Message) collector + Message message = (Message) collector .forChannel(source.output()).poll(1, TimeUnit.SECONDS); User received = mapper.readValue(message.getPayload(), User.class); assertThat( @@ -91,12 +93,12 @@ public class ContentTypeTests { User user = new User("Alice"); String json = mapper.writeValueAsString(user); source.output().send(MessageBuilder.withPayload(user).build()); - Message message = (Message) collector + Message message = (Message) collector .forChannel(source.output()).poll(1, TimeUnit.SECONDS); assertThat( message.getHeaders().get(MessageHeaders.CONTENT_TYPE, MimeType.class) .includes(MimeTypeUtils.APPLICATION_JSON)); - assertThat(json.getBytes()).isEqualTo(message.getPayload()); + assertThat(json).isEqualTo(message.getPayload()); } } @@ -108,17 +110,17 @@ public class ContentTypeTests { MessageCollector collector = context.getBean(MessageCollector.class); Source source = context.getBean(Source.class); source.output().send(MessageBuilder.withPayload("foo").build()); - Message message = (Message) collector + Message message = (Message) collector .forChannel(source.output()).poll(1, TimeUnit.SECONDS); assertThat( message.getHeaders().get(MessageHeaders.CONTENT_TYPE, MimeType.class) .includes(MimeTypeUtils.APPLICATION_JSON)); - assertThat("\"foo\"".getBytes()).isEqualTo(message.getPayload()); + assertThat("foo").isEqualTo(message.getPayload()); } } @Test - public void testSendBynaryDataWithoutContentType() throws Exception { + public void testSendBynaryData() throws Exception { try (ConfigurableApplicationContext context = SpringApplication.run( SourceApplication.class, "--server.port=0", "--spring.jmx.enabled=false")) { @@ -126,7 +128,7 @@ public class ContentTypeTests { MessageCollector collector = context.getBean(MessageCollector.class); Source source = context.getBean(Source.class); byte[] data = new byte[] { 0, 1, 2, 3 }; - source.output().send(MessageBuilder.withPayload(data).build()); + source.output().send(MessageBuilder.withPayload(data).setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_OCTET_STREAM).build()); Message message = (Message) collector .forChannel(source.output()).poll(1, TimeUnit.SECONDS); assertThat( @@ -184,12 +186,11 @@ public class ContentTypeTests { Source source = context.getBean(Source.class); User user = new User("Alice"); source.output().send(MessageBuilder.withPayload(user).build()); - Message message = (Message) collector + Message message = (Message) collector .forChannel(source.output()).poll(1, TimeUnit.SECONDS); assertThat(message.getHeaders().get(MessageHeaders.CONTENT_TYPE, MimeType.class) .includes(MessageConverterUtils.X_JAVA_SERIALIZED_OBJECT)); - ByteArrayInputStream bis = new ByteArrayInputStream((byte[]) (message.getPayload())); - User received = (User) new ObjectInputStream(bis).readObject(); + User received = message.getPayload(); assertThat(user.getName()).isEqualTo(received.getName()); } } @@ -201,15 +202,12 @@ public class ContentTypeTests { "--spring.jmx.enabled=false", "--spring.cloud.stream.bindings.output.contentType=application/x-java-object")) { MessageCollector collector = context.getBean(MessageCollector.class); - Kryo kryo = new Kryo(); Source source = context.getBean(Source.class); User user = new User("Alice"); source.output().send(MessageBuilder.withPayload(user).build()); - Message message = (Message) collector + Message message = (Message) collector .forChannel(source.output()).poll(1, TimeUnit.SECONDS); - com.esotericsoftware.kryo.io.Input input = new com.esotericsoftware.kryo.io.Input(new ByteArrayInputStream(message.getPayload())); - User received = kryo.readObject(input,User.class); - input.close(); + User received = message.getPayload(); assertThat(message.getHeaders().get(MessageHeaders.CONTENT_TYPE, MimeType.class) .includes(MimeType.valueOf(KryoMessageConverter.KRYO_MIME_TYPE))); assertThat(user.getName()).isEqualTo(received.getName()); @@ -227,11 +225,11 @@ public class ContentTypeTests { Source source = context.getBean(Source.class); User user = new User("Alice"); source.output().send(MessageBuilder.withPayload(user).build()); - Message message = (Message) collector + Message message = (Message) collector .forChannel(source.output()).poll(1, TimeUnit.SECONDS); assertThat(message.getHeaders().get(MessageHeaders.CONTENT_TYPE, MimeType.class) .includes(MimeTypeUtils.TEXT_PLAIN)); - assertThat(message.getPayload()).isEqualTo(user.toString().getBytes()); + assertThat(message.getPayload()).isEqualTo(user.toString()); } } @@ -318,7 +316,7 @@ public class ContentTypeTests { } } - @Test + @Test(expected=MessageDeliveryException.class) public void testReceiveKryoWithHeadersOverridingDefault() throws Exception{ try (ConfigurableApplicationContext context = SpringApplication.run( SinkApplication.class, "--server.port=0", @@ -374,7 +372,7 @@ public class ContentTypeTests { @SpringBootApplication public static class SinkApplication { - public LinkedList arguments = new LinkedList(); + public LinkedList arguments = new LinkedList<>(); @StreamListener("POJO_INPUT") public void receive(User user, @Headers Map headers){ @@ -414,6 +412,7 @@ public class ContentTypeTests { } + @SuppressWarnings("serial") public static class User implements Serializable { private String name; diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamEmitterBasicTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamEmitterBasicTests.java index aac728dcc..a62b1ffa1 100644 --- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamEmitterBasicTests.java +++ b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamEmitterBasicTests.java @@ -47,6 +47,7 @@ import static org.assertj.core.api.Assertions.assertThat; * @author Soby Chacko * @author Artem Bilan * @author Vinicius Carvalho + * @author Oleg Zhurakousky */ public class StreamEmitterBasicTests { @@ -125,24 +126,22 @@ public class StreamEmitterBasicTests { context.close(); } - @SuppressWarnings("unchecked") private static void receiveAndValidate(ConfigurableApplicationContext context) throws InterruptedException { Source source = context.getBean(Source.class); MessageCollector messageCollector = context.getBean(MessageCollector.class); - List messages = new ArrayList<>(); + List messages = new ArrayList<>(); for (int i = 0; i < 1000; i++) { - messages.add((byte[]) messageCollector.forChannel(source.output()).poll(5000, TimeUnit.MILLISECONDS).getPayload()); + messages.add((String) messageCollector.forChannel(source.output()).poll(5000, TimeUnit.MILLISECONDS).getPayload()); } for (int i = 0; i < 1000; i++) { assertThat(new String(messages.get(i))).isEqualTo("HELLO WORLD!!" + i); } } - @SuppressWarnings("unchecked") private static void receiveAndValidateMultipleOutputs(ConfigurableApplicationContext context) throws InterruptedException { TestMultiOutboundChannels source = context.getBean(TestMultiOutboundChannels.class); MessageCollector messageCollector = context.getBean(MessageCollector.class); - List messages = new ArrayList<>(); + List messages = new ArrayList<>(); assertMessages(source.output1(), messageCollector, messages); messages.clear(); assertMessages(source.output2(), messageCollector, messages); @@ -151,39 +150,38 @@ public class StreamEmitterBasicTests { messages.clear(); } - @SuppressWarnings("unchecked") private static void receiveAndValidateMultiStreamEmittersInSameContext(ConfigurableApplicationContext context1) throws InterruptedException { TestMultiOutboundChannels source1 = context1.getBean(TestMultiOutboundChannels.class); MessageCollector messageCollector = context1.getBean(MessageCollector.class); - List messages = new ArrayList<>(); + List messages = new ArrayList<>(); assertMessagesX(source1.output1(), messageCollector, messages); messages.clear(); assertMessagesY(source1.output2(), messageCollector, messages); messages.clear(); } - private static void assertMessages(MessageChannel channel, MessageCollector messageCollector, List messages) throws InterruptedException { + private static void assertMessages(MessageChannel channel, MessageCollector messageCollector, List messages) throws InterruptedException { for (int i = 0; i < 1000; i++) { - messages.add((byte[]) messageCollector.forChannel(channel).poll(5000, TimeUnit.MILLISECONDS).getPayload()); + messages.add((String) messageCollector.forChannel(channel).poll(5000, TimeUnit.MILLISECONDS).getPayload()); } for (int i = 0; i < 1000; i++) { assertThat(new String(messages.get(i))).isEqualTo("Hello World!!" + i); } } - private static void assertMessagesX(MessageChannel channel, MessageCollector messageCollector, List messages) throws InterruptedException { + private static void assertMessagesX(MessageChannel channel, MessageCollector messageCollector, List messages) throws InterruptedException { for (int i = 0; i < 1000; i++) { - messages.add((byte[]) messageCollector.forChannel(channel).poll(5000, TimeUnit.MILLISECONDS).getPayload()); + messages.add((String) messageCollector.forChannel(channel).poll(5000, TimeUnit.MILLISECONDS).getPayload()); } for (int i = 0; i < 1000; i++) { assertThat(new String(messages.get(i))).isEqualTo("Hello World!!" + i); } } - private static void assertMessagesY(MessageChannel channel, MessageCollector messageCollector, List messages) throws InterruptedException { + private static void assertMessagesY(MessageChannel channel, MessageCollector messageCollector, List messages) throws InterruptedException { for (int i = 0; i < 1000; i++) { - messages.add((byte[]) messageCollector.forChannel(channel).poll(5000, TimeUnit.MILLISECONDS).getPayload()); + messages.add((String) messageCollector.forChannel(channel).poll(5000, TimeUnit.MILLISECONDS).getPayload()); } for (int i = 0; i < 1000; i++) { assertThat(new String(messages.get(i))).isEqualTo("Hello FooBar!!" + i); diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerGenericFluxInputOutputArgsWithMessageTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerGenericFluxInputOutputArgsWithMessageTests.java index fd3e3826d..77ea29594 100644 --- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerGenericFluxInputOutputArgsWithMessageTests.java +++ b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerGenericFluxInputOutputArgsWithMessageTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -41,11 +41,11 @@ import static org.springframework.cloud.stream.binding.StreamListenerErrorMessag /** * @author Ilayaperumal Gopinathan * @author Vinicius Carvalho + * @author Oleg Zhurakousky */ @SuppressWarnings("unchecked") public class StreamListenerGenericFluxInputOutputArgsWithMessageTests { - @SuppressWarnings("unchecked") private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException { Processor processor = context.getBean(Processor.class); @@ -53,10 +53,10 @@ public class StreamListenerGenericFluxInputOutputArgsWithMessageTests { processor.input().send(MessageBuilder.withPayload(sentPayload) .setHeader("contentType", "text/plain").build()); MessageCollector messageCollector = context.getBean(MessageCollector.class); - Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, + Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); assertThat(result).isNotNull(); - assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase().getBytes()); + assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase()); } @Test diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsTests.java index de8de36a1..eb73f2e1b 100644 --- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsTests.java +++ b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -44,6 +44,7 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Marius Bogoevici * @author Ilayaperumal Gopinathan + * @author Oleg Zhurakousky */ @RunWith(Parameterized.class) public class StreamListenerReactiveInputOutputArgsTests { @@ -55,19 +56,19 @@ public class StreamListenerReactiveInputOutputArgsTests { } @Parameterized.Parameters - public static Collection InputConfigs() { + public static Collection InputConfigs() { return Arrays.asList(new Class[] { ReactorTestInputOutputArgs.class, RxJava1TestInputOutputArgs.class }); } + @SuppressWarnings("unchecked") private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException { - @SuppressWarnings("unchecked") Processor processor = context.getBean(Processor.class); String sentPayload = "hello " + UUID.randomUUID().toString(); processor.input().send(MessageBuilder.withPayload(sentPayload).setHeader("contentType", "text/plain").build()); MessageCollector messageCollector = context.getBean(MessageCollector.class); - Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); + Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); assertThat(result).isNotNull(); - assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase().getBytes()); + assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase()); } @Test diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithMessageTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithMessageTests.java index 0beda9bcf..30cb3a651 100644 --- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithMessageTests.java +++ b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithMessageTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -44,6 +44,7 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Marius Bogoevici * @author Ilayaperumal Gopinathan + * @author Oleg Zhurakousky */ @RunWith(Parameterized.class) public class StreamListenerReactiveInputOutputArgsWithMessageTests { @@ -55,20 +56,20 @@ public class StreamListenerReactiveInputOutputArgsWithMessageTests { } @Parameterized.Parameters - public static Collection InputConfigs() { + public static Collection InputConfigs() { return Arrays.asList(new Class[] { ReactorTestInputOutputArgsWithMessage.class, RxJava1TestInputOutputArgsWithMessage.class }); } + @SuppressWarnings("unchecked") private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException { - @SuppressWarnings("unchecked") Processor processor = context.getBean(Processor.class); String sentPayload = "hello " + UUID.randomUUID().toString(); processor.input().send(MessageBuilder.withPayload(sentPayload).setHeader("contentType", "text/plain").build()); MessageCollector messageCollector = context.getBean(MessageCollector.class); - Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); + Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); assertThat(result).isNotNull(); - assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase().getBytes()); + assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase()); } @Test diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests.java index 79642c890..02587137a 100644 --- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests.java +++ b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -44,6 +44,7 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Marius Bogoevici * @author Ilayaperumal Gopinathan + * @author Oleg Zhurakousky */ @RunWith(Parameterized.class) public class StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests { @@ -55,24 +56,23 @@ public class StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests { } @Parameterized.Parameters - public static Collection InputConfigs() { + public static Collection InputConfigs() { return Arrays.asList(new Class[] { TestInputOutputArgsWithFluxSenderAndFailure.class, TestInputOutputArgsWithObservableSenderAndFailure.class }); } + @SuppressWarnings("unchecked") private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException { - @SuppressWarnings("unchecked") Processor processor = context.getBean(Processor.class); String sentPayload = "hello " + UUID.randomUUID().toString(); processor.input().send(MessageBuilder.withPayload(sentPayload).setHeader("contentType", "text/plain").build()); MessageCollector messageCollector = context.getBean(MessageCollector.class); - Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); + Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); assertThat(result).isNotNull(); - assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase().getBytes()); + assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase()); } private static void sendFailingMessage(ConfigurableApplicationContext context) throws InterruptedException { - @SuppressWarnings("unchecked") Processor processor = context.getBean(Processor.class); processor.input().send(MessageBuilder.withPayload("fail").setHeader("contentType", "text/plain").build()); } diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderTests.java index aa7ce88e0..8312da83c 100644 --- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderTests.java +++ b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -44,6 +44,7 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Marius Bogoevici * @author Ilayaperumal Gopinathan + * @author Oleg Zhurakousky */ @RunWith(Parameterized.class) public class StreamListenerReactiveInputOutputArgsWithSenderTests { @@ -55,20 +56,20 @@ public class StreamListenerReactiveInputOutputArgsWithSenderTests { } @Parameterized.Parameters - public static Collection InputConfigs() { + public static Collection InputConfigs() { return Arrays.asList(new Class[] { ReactorTestInputOutputArgsWithFluxSender.class, RxJava1TestInputOutputArgsWithObservableSender.class }); } + @SuppressWarnings("unchecked") private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException { - @SuppressWarnings("unchecked") Processor processor = context.getBean(Processor.class); String sentPayload = "hello " + UUID.randomUUID().toString(); processor.input().send(MessageBuilder.withPayload(sentPayload).setHeader("contentType", "text/plain").build()); MessageCollector messageCollector = context.getBean(MessageCollector.class); - Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); + Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); assertThat(result).isNotNull(); - assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase().getBytes()); + assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase()); } @Test diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveMethodWithReturnTypeTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveMethodWithReturnTypeTests.java index ac8127e95..d5b4b0379 100644 --- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveMethodWithReturnTypeTests.java +++ b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveMethodWithReturnTypeTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -45,6 +45,7 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Marius Bogoevici * @author Ilayaperumal Gopinathan + * @author Oleg Zhurakousky */ @RunWith(Parameterized.class) public class StreamListenerReactiveMethodWithReturnTypeTests { @@ -56,22 +57,22 @@ public class StreamListenerReactiveMethodWithReturnTypeTests { } @Parameterized.Parameters - public static Collection InputConfigs() { + public static Collection InputConfigs() { return Arrays.asList(new Class[] { ReactorTestReturn1.class, ReactorTestReturn2.class, ReactorTestReturn3.class, ReactorTestReturn4.class, RxJava1TestReturn1.class, RxJava1TestReturn2.class, RxJava1TestReturn3.class, RxJava1TestReturn4.class }); } + @SuppressWarnings("unchecked") private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException { - @SuppressWarnings("unchecked") Processor processor = context.getBean(Processor.class); String sentPayload = "hello " + UUID.randomUUID().toString(); processor.input().send(MessageBuilder.withPayload(sentPayload).setHeader("contentType", "text/plain").build()); MessageCollector messageCollector = context.getBean(MessageCollector.class); - Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); + Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); assertThat(result).isNotNull(); - assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase().getBytes()); + assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase()); } @Test diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithFailureTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithFailureTests.java index 369c683f4..6b97702cf 100644 --- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithFailureTests.java +++ b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithFailureTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -45,6 +45,7 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Marius Bogoevici * @author Ilayaperumal Gopinathan + * @author Oleg Zhurakousky */ @RunWith(Parameterized.class) public class StreamListenerReactiveReturnWithFailureTests { @@ -56,7 +57,7 @@ public class StreamListenerReactiveReturnWithFailureTests { } @Parameterized.Parameters - public static Collection InputConfigs() { + public static Collection InputConfigs() { return Arrays.asList(new Class[] { ReactorTestReturnWithFailure1.class, ReactorTestReturnWithFailure2.class, ReactorTestReturnWithFailure3.class, ReactorTestReturnWithFailure4.class, RxJava1TestReturnWithFailure1.class, @@ -64,19 +65,18 @@ public class StreamListenerReactiveReturnWithFailureTests { RxJava1TestReturnWithFailure4.class }); } + @SuppressWarnings("unchecked") private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException { - @SuppressWarnings("unchecked") Processor processor = context.getBean(Processor.class); String sentPayload = "hello " + UUID.randomUUID().toString(); processor.input().send(MessageBuilder.withPayload(sentPayload).setHeader("contentType", "text/plain").build()); MessageCollector messageCollector = context.getBean(MessageCollector.class); - Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); + Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); assertThat(result).isNotNull(); - assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase().getBytes()); + assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase()); } private static void sendFailingMessage(ConfigurableApplicationContext context) throws InterruptedException { - @SuppressWarnings("unchecked") Processor processor = context.getBean(Processor.class); processor.input().send(MessageBuilder.withPayload("fail").setHeader("contentType", "text/plain").build()); } diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithMessageTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithMessageTests.java index eb4796ad0..cc8438c17 100644 --- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithMessageTests.java +++ b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithMessageTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -45,6 +45,7 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Marius Bogoevici * @author Ilayaperumal Gopinathan + * @author Oleg Zhurakousky */ @RunWith(Parameterized.class) public class StreamListenerReactiveReturnWithMessageTests { @@ -56,7 +57,7 @@ public class StreamListenerReactiveReturnWithMessageTests { } @Parameterized.Parameters - public static Collection InputConfigs() { + public static Collection InputConfigs() { return Arrays.asList(new Class[] { ReactorTestReturnWithMessage1.class, ReactorTestReturnWithMessage2.class, ReactorTestReturnWithMessage3.class, ReactorTestReturnWithMessage4.class, RxJava1TestReturnWithMessage1.class, @@ -64,15 +65,15 @@ public class StreamListenerReactiveReturnWithMessageTests { RxJava1TestReturnWithMessage4.class }); } + @SuppressWarnings("unchecked") private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException { - @SuppressWarnings("unchecked") Processor processor = context.getBean(Processor.class); String sentPayload = "hello " + UUID.randomUUID().toString(); processor.input().send(MessageBuilder.withPayload(sentPayload).setHeader("contentType", "text/plain").build()); MessageCollector messageCollector = context.getBean(MessageCollector.class); - Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); + Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); assertThat(result).isNotNull(); - assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase().getBytes()); + assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase()); } @Test diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithPojoTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithPojoTests.java index ea2635c3b..7ec73fe78 100644 --- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithPojoTests.java +++ b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithPojoTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -47,6 +47,7 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Marius Bogoevici * @author Ilayaperumal Gopinathan + * @author Oleg Zhurakousky */ @RunWith(Parameterized.class) public class StreamListenerReactiveReturnWithPojoTests { @@ -60,24 +61,23 @@ public class StreamListenerReactiveReturnWithPojoTests { } @Parameterized.Parameters - public static Collection InputConfigs() { + public static Collection InputConfigs() { return Arrays.asList(new Class[] { ReactorTestReturnWithPojo1.class, ReactorTestReturnWithPojo2.class, ReactorTestReturnWithPojo3.class, ReactorTestReturnWithPojo4.class, RxJava1TestReturnWithPojo1.class, RxJava1TestReturnWithPojo2.class, RxJava1TestReturnWithPojo3.class, RxJava1TestReturnWithPojo4.class }); } @Test + @SuppressWarnings("unchecked") public void testReturnWithPojo() throws Exception { ConfigurableApplicationContext context = SpringApplication.run(this.configClass, "--server.port=0", "--spring.jmx.enabled=false"); - @SuppressWarnings("unchecked") Processor processor = context.getBean(Processor.class); processor.input().send(MessageBuilder.withPayload("{\"message\":\"helloPojo\"}") .setHeader("contentType", "application/json").build()); MessageCollector messageCollector = context.getBean(MessageCollector.class); - Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); + Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); assertThat(result).isNotNull(); - assertThat(result.getPayload()).isInstanceOf(byte[].class); BarPojo barPojo = mapper.readValue(result.getPayload(),BarPojo.class); assertThat(barPojo.getBarMessage()).isEqualTo("helloPojo"); context.close(); diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerWildCardFluxInputOutputArgsWithMessageTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerWildCardFluxInputOutputArgsWithMessageTests.java index 1d86c3f47..f0a813994 100644 --- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerWildCardFluxInputOutputArgsWithMessageTests.java +++ b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerWildCardFluxInputOutputArgsWithMessageTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -42,18 +42,19 @@ import static org.springframework.cloud.stream.binding.StreamListenerErrorMessag /** * @author Ilayaperumal Gopinathan + * @author Oleg Zhurakousky */ public class StreamListenerWildCardFluxInputOutputArgsWithMessageTests { + @SuppressWarnings("unchecked") private static void sendMessageAndValidate(ConfigurableApplicationContext context) throws InterruptedException { - @SuppressWarnings("unchecked") Processor processor = context.getBean(Processor.class); String sentPayload = "hello " + UUID.randomUUID().toString(); processor.input().send(MessageBuilder.withPayload(sentPayload).setHeader("contentType", "text/plain").build()); MessageCollector messageCollector = context.getBean(MessageCollector.class); - Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); + Message result = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); assertThat(result).isNotNull(); - assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase().getBytes()); + assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase()); } @Test diff --git a/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/test/binder/TestSupportBinder.java b/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/test/binder/TestSupportBinder.java index c52e1fc41..a165e012c 100644 --- a/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/test/binder/TestSupportBinder.java +++ b/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/test/binder/TestSupportBinder.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-2017 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. @@ -27,7 +27,9 @@ import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.ConsumerProperties; import org.springframework.cloud.stream.binder.ProducerProperties; +import org.springframework.cloud.stream.binding.MessageConverterConfigurer; import org.springframework.cloud.stream.test.matcher.MessageQueueMatcher; +import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -47,6 +49,7 @@ import org.springframework.util.Assert; * @author Eric Bottard * @author Gary Russell * @author Mark Fisher + * @author Oleg Zhurakousky * @see MessageQueueMatcher */ public class TestSupportBinder implements Binder { @@ -97,6 +100,8 @@ public class TestSupportBinder implements Binder>> results = new HashMap<>(); private BlockingQueue> register(MessageChannel channel) { + // we need to add this intercepter to ensure MessageCollector's compatibility with previous versions of SCSt + ((AbstractMessageChannel)channel).addInterceptor(new MessageConverterConfigurer.InboundMessageConvertingInterceptor()); LinkedBlockingDeque> result = new LinkedBlockingDeque<>(); Assert.isTrue(!results.containsKey(channel), "Channel [" + channel + "] was already bound"); results.put(channel, result); diff --git a/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/aggregate/bean/AggregateWithBeanTest.java b/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/aggregate/bean/AggregateWithBeanTest.java index 2e4c619cd..ee3097654 100644 --- a/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/aggregate/bean/AggregateWithBeanTest.java +++ b/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/aggregate/bean/AggregateWithBeanTest.java @@ -52,13 +52,14 @@ public class AggregateWithBeanTest { public AggregateApplication aggregateApplication; @Test + @SuppressWarnings("unchecked") public void testAggregateApplication() throws InterruptedException { Processor uppercaseProcessor = aggregateApplication.getBinding(Processor.class, "upper"); Processor suffixProcessor = aggregateApplication.getBinding(Processor.class, "suffix"); uppercaseProcessor.input().send(MessageBuilder.withPayload("Hello").build()); - Message receivedMessage = (Message) messageCollector.forChannel(suffixProcessor.output()).poll(1, TimeUnit.SECONDS); + Message receivedMessage = (Message) messageCollector.forChannel(suffixProcessor.output()).poll(1, TimeUnit.SECONDS); assertThat(receivedMessage).isNotNull(); - assertThat(receivedMessage.getPayload()).isEqualTo("HELLO WORLD!".getBytes()); + assertThat(receivedMessage.getPayload()).isEqualTo("HELLO WORLD!"); } @SpringBootApplication diff --git a/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/aggregate/main/AggregateWithMainTest.java b/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/aggregate/main/AggregateWithMainTest.java index 6cddc9fdf..8f0fbe9d5 100644 --- a/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/aggregate/main/AggregateWithMainTest.java +++ b/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/aggregate/main/AggregateWithMainTest.java @@ -42,6 +42,7 @@ import static org.assertj.core.api.Assertions.assertThat; */ public class AggregateWithMainTest { + @SuppressWarnings("unchecked") @Test public void testAggregateApplication() throws InterruptedException { // emulate a main method @@ -55,9 +56,9 @@ public class AggregateWithMainTest { Processor uppercaseProcessor = aggregateAccessor.getBinding(Processor.class, "upper"); Processor suffixProcessor = aggregateAccessor.getBinding(Processor.class, "suffix"); uppercaseProcessor.input().send(MessageBuilder.withPayload("Hello").build()); - Message receivedMessage = (Message) messageCollector.forChannel(suffixProcessor.output()).poll(1, TimeUnit.SECONDS); + Message receivedMessage = (Message) messageCollector.forChannel(suffixProcessor.output()).poll(1, TimeUnit.SECONDS); assertThat(receivedMessage).isNotNull(); - assertThat(receivedMessage.getPayload()).isEqualTo("HELLO WORLD!".getBytes()); + assertThat(receivedMessage.getPayload()).isEqualTo("HELLO WORLD!"); context.close(); } diff --git a/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/disable/AutoconfigurationDisabledTest.java b/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/disable/AutoconfigurationDisabledTest.java index 7d12e3a44..9448b849a 100644 --- a/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/disable/AutoconfigurationDisabledTest.java +++ b/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/disable/AutoconfigurationDisabledTest.java @@ -55,13 +55,14 @@ public class AutoconfigurationDisabledTest { @Autowired public Processor processor; + @SuppressWarnings("unchecked") @Test public void testAutoconfigurationDisabled() throws Exception { processor.input().send(MessageBuilder.withPayload("Hello").build()); // Since the interaction is synchronous, the result should be immediate - Message response = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); + Message response = (Message) messageCollector.forChannel(processor.output()).poll(1000, TimeUnit.MILLISECONDS); assertThat(response).isNotNull(); - assertThat(response.getPayload()).isEqualTo("Hello world".getBytes()); + assertThat(response.getPayload()).isEqualTo("Hello world"); } @SpringBootApplication(exclude = TestSupportBinderAutoConfiguration.class) diff --git a/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/example/ExampleTest.java b/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/example/ExampleTest.java index 693d9afd3..08444b8de 100644 --- a/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/example/ExampleTest.java +++ b/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/example/ExampleTest.java @@ -22,9 +22,7 @@ import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.cloud.stream.annotation.Bindings; import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.binder.BinderFactory; import org.springframework.cloud.stream.messaging.Processor; import org.springframework.cloud.stream.test.binder.MessageCollector; import org.springframework.integration.annotation.Transformer; @@ -45,14 +43,10 @@ import static org.assertj.core.api.Assertions.assertThat; @DirtiesContext public class ExampleTest { - @Autowired - private BinderFactory binderFactory; - @Autowired private MessageCollector messageCollector; @Autowired - @Bindings(MyProcessor.class) private Processor processor; @Test @@ -60,8 +54,8 @@ public class ExampleTest { public void testWiring() { Message message = new GenericMessage<>("hello"); this.processor.input().send(message); - Message received = (Message) this.messageCollector.forChannel(this.processor.output()).poll(); - assertThat(received.getPayload()).isEqualTo("hello world".getBytes()); + Message received = (Message) this.messageCollector.forChannel(this.processor.output()).poll(); + assertThat(received.getPayload()).isEqualTo("hello world"); } @SpringBootApplication diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/MessageSerializationUtils.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/MessageSerializationUtils.java index 0f4eb6220..d7ea670b8 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/MessageSerializationUtils.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/MessageSerializationUtils.java @@ -16,30 +16,54 @@ package org.springframework.cloud.stream.binder; +import java.nio.charset.StandardCharsets; + import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; +import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; +import org.springframework.util.StringUtils; /** * Utility class for serializing and de-serializing the message payload. * * @author Soby Chacko * @author Vinicius Carvalho + * @author Oleg Zhurakousky */ +@Deprecated public abstract class MessageSerializationUtils { /** * Serialize the message payload unless it is a byte array. * * @param message the message with the payload to serialize - * @return the Message with teh serialized payload + * @return the Message with the serialized payload */ public static MessageValues serializePayload(Message message) { Object originalPayload = message.getPayload(); + boolean setOriginalContentType = (originalPayload instanceof String); + Assert.isTrue(originalPayload instanceof byte[] || originalPayload instanceof String, + "Failed to convert message's payload. No suitable converter found for provided contentType: " + + message.getHeaders().get(MessageHeaders.CONTENT_TYPE) + " and paylod: " + originalPayload); Object originalContentType = message.getHeaders().get(MessageHeaders.CONTENT_TYPE); + // Pass content type as String since some transport adapters will exclude + // CONTENT_TYPE Header otherwise + String contentType = null; + if (originalContentType != null) { + contentType = setOriginalContentType ? JavaClassMimeTypeUtils.mimeTypeFromObject(originalPayload, + ObjectUtils.nullSafeToString(originalContentType)).toString() : originalContentType.toString() ; + } + + Object payload = originalPayload instanceof byte[] ? originalPayload : ((String) originalPayload).getBytes(StandardCharsets.UTF_8); MessageValues messageValues = new MessageValues(message); - messageValues.setPayload(originalPayload); - messageValues.put(MessageHeaders.CONTENT_TYPE, originalContentType); + messageValues.setPayload(payload); + if (StringUtils.hasText(contentType)) { + messageValues.put(MessageHeaders.CONTENT_TYPE, contentType); + if (originalContentType != null && !originalContentType.toString().equals(contentType.toString())) { + messageValues.put(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE, originalContentType.toString()); + } + } return messageValues; } - } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java index 0772e9380..3cbfb4aef 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java @@ -16,6 +16,8 @@ package org.springframework.cloud.stream.binding; +import java.nio.charset.StandardCharsets; + import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; @@ -39,14 +41,15 @@ import org.springframework.integration.support.MutableMessageHeaders; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHeaders; -import org.springframework.messaging.converter.AbstractMessageConverter; -import org.springframework.messaging.converter.MessageConversionException; +import org.springframework.messaging.converter.DefaultContentTypeResolver; import org.springframework.messaging.converter.MessageConverter; import org.springframework.messaging.support.ChannelInterceptorAdapter; import org.springframework.messaging.support.ErrorMessage; +import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; import org.springframework.util.MimeType; +import org.springframework.util.MimeTypeUtils; import org.springframework.util.StringUtils; /** @@ -61,6 +64,7 @@ import org.springframework.util.StringUtils; * @author Maxim Kirilov * @author Gary Russell * @author Soby Chacko + * @author Oleg Zhurakousky */ public class MessageConverterConfigurer implements MessageChannelConfigurer, BeanFactoryAware, InitializingBean { @@ -114,7 +118,7 @@ public class MessageConverterConfigurer AbstractMessageChannel messageChannel = (AbstractMessageChannel) channel; final BindingProperties bindingProperties = this.bindingServiceProperties .getBindingProperties(channelName); - final String contentType = bindingProperties.getContentType(); + String contentType = bindingProperties.getContentType(); ProducerProperties producerProperties = bindingProperties.getProducer(); if (!input && producerProperties != null && producerProperties.isPartitioned()) { messageChannel.addInterceptor(new PartitioningInterceptor(bindingProperties, @@ -122,7 +126,7 @@ public class MessageConverterConfigurer getPartitionSelectorStrategy(producerProperties))); } if (input) { - messageChannel.addInterceptor(new LegacyContentTypeHeaderInterceptor()); + messageChannel.addInterceptor(new InboundMessageConvertingInterceptor()); } // TODO: Set all interceptors in the correct order for input/output channels if (StringUtils.hasText(contentType)) { @@ -194,7 +198,6 @@ public class MessageConverterConfigurer } return Math.abs(hashCode); } - } private final class ContentTypeConvertingInterceptor @@ -206,15 +209,12 @@ public class MessageConverterConfigurer private final MessageConverter messageConverter; - private final boolean provideHint; - private ContentTypeConvertingInterceptor(String contentType, boolean input) { this.mimeType = MessageConverterUtils.getMimeType(contentType); this.input = input; this.messageConverter = MessageConverterConfigurer.this.compositeMessageConverterFactory .getMessageConverterForAllRegistered(); - this.provideHint = this.messageConverter instanceof AbstractMessageConverter; } @Override @@ -224,7 +224,7 @@ public class MessageConverterConfigurer return message; } - Message sentMessage = null; + Message sentMessage = message; Object converted; // bypass conversion for raw bytes or input channels if (this.input || message.getPayload() instanceof byte[]) { @@ -254,14 +254,6 @@ public class MessageConverterConfigurer .build(); } } - if (sentMessage == null) { - throw new MessageConversionException(message, - this.messageConverter.getClass().toString() - + " could not convert '" + message - + "' to the configured output type: '" + this.mimeType - + "'"); - - } return sentMessage; } } @@ -302,20 +294,77 @@ public class MessageConverterConfigurer } } - private final class LegacyContentTypeHeaderInterceptor extends ChannelInterceptorAdapter { + public final static class InboundMessageConvertingInterceptor extends ChannelInterceptorAdapter { + + private final DefaultContentTypeResolver contentTypeResolver = new DefaultContentTypeResolver(); + private final CompositeMessageConverterFactory converterFactory = new CompositeMessageConverterFactory(); @Override public Message preSend(Message message, MessageChannel channel) { - if (!message.getHeaders().containsKey(BinderHeaders.SCST_VERSION) || - !message.getHeaders().get(BinderHeaders.SCST_VERSION).equals("2.x")) { - Object originalContentType = message.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE); - return originalContentType != null ? MessageConverterConfigurer.this.messageBuilderFactory - .fromMessage(message) - .setHeader(MessageHeaders.CONTENT_TYPE, originalContentType).build() : message; + Class targetClass = null; + MessageConverter converter = null; + MimeType contentType = message.getHeaders().containsKey(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE) + ? MimeType.valueOf((String)message.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)) + : contentTypeResolver.resolve(message.getHeaders()); + + if (contentType != null){ + if (equalTypeAndSubType(MessageConverterUtils.X_JAVA_SERIALIZED_OBJECT, contentType) || + equalTypeAndSubType(MessageConverterUtils.X_JAVA_OBJECT, contentType)) { + // for Java and Kryo de-serialization we need to reset the content type + message = MessageBuilder.fromMessage(message).setHeader(MessageHeaders.CONTENT_TYPE, contentType).build(); + converter = equalTypeAndSubType(MessageConverterUtils.X_JAVA_SERIALIZED_OBJECT, contentType) + ? converterFactory.getMessageConverterForType(contentType) + : converterFactory.getMessageConverterForAllRegistered(); + String targetClassName = contentType.getParameter("type"); + if (StringUtils.hasText(targetClassName)) { + try { + targetClass = Class.forName(targetClassName, false, Thread.currentThread().getContextClassLoader()); + } + catch (Exception e) { + throw new IllegalStateException("Failed to determine class name for contentType: " + + message.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE), e); + } + } + } } + + Object payload; + if (converter != null){ + Assert.isTrue(!(equalTypeAndSubType(MessageConverterUtils.X_JAVA_OBJECT, contentType) && targetClass == null), + "Can not deserialize into message since 'contentType` has not " + + "being encoded with the actual target type." + + "Consider 'application/x-java-object; type=foo.bar.MyClass'"); + payload = converter.fromMessage(message, targetClass); + } + else { + MimeType deserializeContentType = contentTypeResolver.resolve(message.getHeaders()); + if (deserializeContentType == null) { + deserializeContentType = contentType; + } + payload = deserializeContentType == null ? message.getPayload() : this.deserializePayload(message.getPayload(), deserializeContentType); + } + message = MessageBuilder.withPayload(payload) + .copyHeaders(message.getHeaders()) + .setHeader(MessageHeaders.CONTENT_TYPE, contentType) + .removeHeader(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE) + .build(); return message; } + private Object deserializePayload(Object payload, MimeType contentType) { + if (payload instanceof byte[] && ("text".equalsIgnoreCase(contentType.getType()) || + equalTypeAndSubType(MimeTypeUtils.APPLICATION_JSON, contentType))) { + payload = new String((byte[])payload, StandardCharsets.UTF_8); + } + return payload; + } + } + + /* + * Candidate to go into some utils class + */ + private static boolean equalTypeAndSubType(MimeType m1, MimeType m2) { + return m1 != null && m2 != null && m1.getType().equalsIgnoreCase(m2.getType()) && m1.getSubtype().equalsIgnoreCase(m2.getSubtype()); } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerAnnotationBeanPostProcessor.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerAnnotationBeanPostProcessor.java index 99edea991..38ae2e617 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerAnnotationBeanPostProcessor.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerAnnotationBeanPostProcessor.java @@ -71,6 +71,7 @@ import org.springframework.util.StringUtils; * @author Marius Bogoevici * @author Ilayaperumal Gopinathan * @author Soby Chacko + * @author Oleg Zhurakousky */ public class StreamListenerAnnotationBeanPostProcessor implements BeanPostProcessor, ApplicationContextAware, BeanFactoryAware, SmartInitializingSingleton, @@ -198,7 +199,7 @@ public class StreamListenerAnnotationBeanPostProcessor String methodAnnotatedOutboundName) { int methodArgumentsLength = method.getParameterTypes().length; for (int parameterIndex = 0; parameterIndex < methodArgumentsLength; parameterIndex++) { - MethodParameter methodParameter = MethodParameter.forMethodOrConstructor(method, parameterIndex); + MethodParameter methodParameter = MethodParameter.forExecutable(method, parameterIndex); if (methodParameter.hasParameterAnnotation(Input.class)) { String inboundName = (String) AnnotationUtils .getValue(methodParameter.getParameterAnnotation(Input.class)); @@ -260,7 +261,7 @@ public class StreamListenerAnnotationBeanPostProcessor String outboundName) { Object[] arguments = new Object[method.getParameterTypes().length]; for (int parameterIndex = 0; parameterIndex < arguments.length; parameterIndex++) { - MethodParameter methodParameter = MethodParameter.forMethodOrConstructor(method, parameterIndex); + MethodParameter methodParameter = MethodParameter.forExecutable(method, parameterIndex); Class parameterType = methodParameter.getParameterType(); Object targetReferenceValue = null; if (methodParameter.hasParameterAnnotation(Input.class)) { @@ -300,8 +301,7 @@ public class StreamListenerAnnotationBeanPostProcessor Object result = method.invoke(bean, arguments); if (!StringUtils.hasText(outboundName)) { for (int parameterIndex = 0; parameterIndex < method.getParameterTypes().length; parameterIndex++) { - MethodParameter methodParameter = MethodParameter.forMethodOrConstructor(method, - parameterIndex); + MethodParameter methodParameter = MethodParameter.forExecutable(method, parameterIndex); if (methodParameter.hasParameterAnnotation(Output.class)) { outboundName = methodParameter.getParameterAnnotation(Output.class).value(); } @@ -390,6 +390,7 @@ public class StreamListenerAnnotationBeanPostProcessor handler.setApplicationContext(this.applicationContext); handler.setChannelResolver(this.binderAwareChannelResolver); handler.afterPropertiesSet(); + this.applicationContext.getBeanFactory().registerSingleton(handler.getClass().getSimpleName() + handler.hashCode(), handler); applicationContext.getBean(mappedBindingEntry.getKey(), SubscribableChannel.class).subscribe(handler); } this.mappedListenerMethods.clear(); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java index c0f880313..245383806 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-2017 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. @@ -23,7 +23,6 @@ import com.fasterxml.jackson.annotation.JsonInclude.Include; import org.springframework.cloud.stream.binder.ConsumerProperties; import org.springframework.cloud.stream.binder.ProducerProperties; -import org.springframework.util.MimeTypeUtils; import org.springframework.validation.annotation.Validated; /** @@ -32,6 +31,7 @@ import org.springframework.validation.annotation.Validated; * @author Ilayaperumal Gopinathan * @author Gary Russell * @author Soby Chacko + * @author Oleg Zhurakousky */ @JsonInclude(Include.NON_DEFAULT) @Validated @@ -56,7 +56,7 @@ public class BindingProperties { // Properties for both inbound/outbound - private String contentType = MimeTypeUtils.APPLICATION_JSON_VALUE; + private String contentType = "application/json"; private String binder; @@ -117,6 +117,7 @@ public class BindingProperties { return consumer == null || producer == null; } + @Override public String toString() { StringBuilder sb = new StringBuilder(); sb.append("destination=" + this.destination); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java index 1b4a93bbc..b78a65422 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java @@ -16,17 +16,24 @@ package org.springframework.cloud.stream.converter; +import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Collections; import java.util.List; import com.fasterxml.jackson.databind.ObjectMapper; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.AbstractMessageConverter; import org.springframework.messaging.converter.ByteArrayMessageConverter; import org.springframework.messaging.converter.CompositeMessageConverter; +import org.springframework.messaging.converter.MappingJackson2MessageConverter; +import org.springframework.messaging.converter.MessageConversionException; import org.springframework.messaging.converter.MessageConverter; import org.springframework.util.CollectionUtils; import org.springframework.util.MimeType; @@ -38,6 +45,7 @@ import org.springframework.util.MimeType; * @author Ilayaperumal Gopinathan * @author Marius Bogoevici * @author Vinicius Carvalho + * @author Oleg Zhurakousky */ public class CompositeMessageConverterFactory { @@ -69,7 +77,34 @@ public class CompositeMessageConverterFactory { private void initDefaultConverters() { this.converters.add(new TupleJsonMessageConverter(this.objectMapper)); - CustomJackson2MappingMessageConverter jsonMessageConverter = new CustomJackson2MappingMessageConverter(); + MappingJackson2MessageConverter jsonMessageConverter = new MappingJackson2MessageConverter() { + @Override + protected Object convertToInternal(Object payload, @Nullable MessageHeaders headers, @Nullable Object conversionHint) { + if (payload instanceof byte[]){ + return payload; + } + else if (payload instanceof String) { + return ((String)payload).getBytes(StandardCharsets.UTF_8); + } + else { + return super.convertToInternal(payload, headers, conversionHint); + } + } + + @Override + protected Object convertFromInternal(Message message, Class targetClass, @Nullable Object conversionHint) { + try{ + return super.convertFromInternal(message, targetClass, conversionHint); + } catch (MessageConversionException me){ + //Strings need special treatment + if(targetClass.isAssignableFrom(String.class)){ + return message.getPayload(); + } + throw me; + } + } + }; + jsonMessageConverter.setStrictContentTypeMatch(true); if (this.objectMapper != null) { jsonMessageConverter.setObjectMapper(this.objectMapper); } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CustomJackson2MappingMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CustomJackson2MappingMessageConverter.java deleted file mode 100644 index 769411c03..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CustomJackson2MappingMessageConverter.java +++ /dev/null @@ -1,54 +0,0 @@ -/* - * Copyright 2017 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. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.stream.converter; - -import org.springframework.lang.Nullable; -import org.springframework.messaging.Message; -import org.springframework.messaging.converter.MappingJackson2MessageConverter; -import org.springframework.messaging.converter.MessageConversionException; - -/** - * Custom implementation of {@link org.springframework.messaging.converter.MappingJackson2MessageConverter} that handles special String cases. - * If the target of conversion is a String, it tries to read it from a quoted json string, to properly remove the quotes, if it fails, it then just - * returns the original string as it is just a raw json string needed for the target. - * - * @author Vinicius Carvalho - */ - -public class CustomJackson2MappingMessageConverter extends MappingJackson2MessageConverter{ - - public CustomJackson2MappingMessageConverter() { - super(); - setSerializedPayloadClass(byte[].class); - setStrictContentTypeMatch(true); - } - - @Override - protected Object convertFromInternal(Message message, Class targetClass, @Nullable Object conversionHint) { - try{ - return super.convertFromInternal(message, targetClass, conversionHint); - }catch (MessageConversionException me){ - //Strings need special treatment - if(targetClass.isAssignableFrom(String.class)){ - return message.getPayload(); - } - throw me; - } - - } - -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JavaSerializationMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JavaSerializationMessageConverter.java index 61e078422..636e9f130 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JavaSerializationMessageConverter.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JavaSerializationMessageConverter.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -30,6 +30,7 @@ import org.springframework.messaging.converter.AbstractMessageConverter; /** * @author Marius Bogoevici + * @author Oleg Zhurakousky */ public class JavaSerializationMessageConverter extends AbstractMessageConverter { @@ -39,7 +40,10 @@ public class JavaSerializationMessageConverter extends AbstractMessageConverter @Override protected boolean supports(Class clazz) { - return Serializable.class.isAssignableFrom(clazz); + if (clazz != null){ + return Serializable.class.isAssignableFrom(clazz); + } + return true; } @Override diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageConverterConfigurerTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageConverterConfigurerTests.java index b479d02dd..ad750b5bd 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageConverterConfigurerTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageConverterConfigurerTests.java @@ -18,6 +18,7 @@ package org.springframework.cloud.stream.binding; import java.util.Collections; +import org.junit.Ignore; import org.junit.Test; import org.springframework.cloud.stream.binder.BinderHeaders; @@ -39,15 +40,14 @@ import static org.junit.Assert.fail; /** * @author Gary Russell + * @author Oleg Zhurakousky * @since 1.3 * */ public class MessageConverterConfigurerTests { - /** - * @since 2.0 bad contentType will result in MessageConversionException - */ - @Test(expected = MessageConversionException.class) + + @Test public void testConfigureOutputChannelWithBadContentType() { BindingServiceProperties props = new BindingServiceProperties(); BindingProperties bindingProps = new BindingProperties(); @@ -62,10 +62,11 @@ public class MessageConverterConfigurerTests { Collections. singletonMap(MessageHeaders.CONTENT_TYPE, "bad/ct"))); Message received = out.receive(0); assertThat(received).isNotNull(); - assertThat(received.getPayload()).isEqualTo("{\"bar\":\"bar\"}"); + assertThat(received.getPayload()).isInstanceOf(Foo.class); } @Test + @Ignore public void testConfigureOutputChannelCannotConvert() { BindingServiceProperties props = new BindingServiceProperties(); BindingProperties bindingProps = new BindingProperties(); @@ -119,7 +120,7 @@ public class MessageConverterConfigurerTests { Message received = in.receive(0); assertThat(received).isNotNull(); assertThat(received.getPayload()).isEqualTo(foo); - assertThat(received.getHeaders().get(MessageHeaders.CONTENT_TYPE)).isEqualTo("application/json"); + assertThat(received.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString()).isEqualTo("application/json"); } public static class Foo { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/converter/CustomMappingJackson2MessageConverterTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/converter/CustomMappingJackson2MessageConverterTests.java deleted file mode 100644 index 1b346164f..000000000 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/converter/CustomMappingJackson2MessageConverterTests.java +++ /dev/null @@ -1,63 +0,0 @@ -/* - * Copyright 2017 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. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.stream.converter; - -import java.util.Map; - -import com.fasterxml.jackson.databind.ObjectMapper; -import org.junit.Test; - -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHeaders; -import org.springframework.messaging.support.MessageBuilder; -import static org.assertj.core.api.Assertions.assertThat; - -/** - * @author Vinicius Carvalho - */ -public class CustomMappingJackson2MessageConverterTests { - - @Test - public void convertFromStructuredJsonIntoPojo() throws Exception { - String payload = "{\"id\":1}"; - Message message = MessageBuilder.withPayload(payload.getBytes()).setHeader(MessageHeaders.CONTENT_TYPE,"application/json").build(); - CustomJackson2MappingMessageConverter converter = new CustomJackson2MappingMessageConverter(); - Object converted = converter.convertFromInternal(message, Map.class,null); - assertThat(((Map)converted).get("id")).isNotNull(); - } - - @Test - public void convertFromStructuredJsonIntoString() throws Exception { - String payload = "{\"id\":1}"; - Message message = MessageBuilder.withPayload(payload.getBytes()).setHeader(MessageHeaders.CONTENT_TYPE,"application/json").build(); - CustomJackson2MappingMessageConverter converter = new CustomJackson2MappingMessageConverter(); - Object converted = converter.convertFromInternal(message, String.class,null); - assertThat(converted).isNotNull(); - assertThat(payload).isEqualTo(new String((byte[])converted)); - } - - @Test - public void convertFromJsonString() throws Exception { - ObjectMapper mapper = new ObjectMapper(); - String payload = mapper.writeValueAsString("foo"); - Message message = MessageBuilder.withPayload(payload.getBytes()).setHeader(MessageHeaders.CONTENT_TYPE,"application/json").build(); - CustomJackson2MappingMessageConverter converter = new CustomJackson2MappingMessageConverter(); - Object converted = converter.convertFromInternal(message, String.class,null); - assertThat(converted).isNotNull(); - assertThat("foo").isEqualTo((String)converted); - } -}