diff --git a/pom.xml b/pom.xml index e92f5af1e..063ae0d77 100644 --- a/pom.xml +++ b/pom.xml @@ -26,7 +26,6 @@ 1.0.0.RELEASE 1.0.0.RELEASE Californium-SR5 - 3.0.3 2.1 2.1.0.RELEASE @@ -81,11 +80,6 @@ spring-cloud-stream-test-support-internal ${project.version} - - com.esotericsoftware - kryo-shaded - ${kryo-shaded.version} - io.projectreactor reactor-bom diff --git a/spring-cloud-stream-binder-test/pom.xml b/spring-cloud-stream-binder-test/pom.xml index 5280dda63..96957a779 100644 --- a/spring-cloud-stream-binder-test/pom.xml +++ b/spring-cloud-stream-binder-test/pom.xml @@ -27,6 +27,10 @@ + + com.fasterxml.jackson.core + jackson-databind + org.springframework spring-web 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 75292e929..27ec74678 100644 --- a/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java +++ b/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java @@ -39,8 +39,6 @@ import org.springframework.cloud.stream.binding.StreamListenerMessageHandler; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; -import org.springframework.cloud.stream.converter.JavaSerializationMessageConverter; -import org.springframework.cloud.stream.converter.KryoMessageConverter; import org.springframework.cloud.stream.converter.MessageConverterUtils; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.Lifecycle; @@ -217,114 +215,6 @@ public abstract class AbstractBinderTests producerBinding = binder.bindProducer( - String.format("foo%s0x", getDestinationNameDelimiter()), - moduleOutputChannel, outputBindingProperties.getProducer()); - Binding consumerBinding = binder.bindConsumer( - String.format("foo%s0x", getDestinationNameDelimiter()), - "testSendAndReceiveKryo", moduleInputChannel, - inputBindingProperties.getConsumer()); - 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(message1 -> { - try { - inboundMessageRef.set((Message) message1); - } - finally { - latch.countDown(); - } - }); - - moduleOutputChannel.send(message); - Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message"); - - KryoMessageConverter kryo = new KryoMessageConverter(null, true); - Foo fooPayload = (Foo) kryo.fromMessage(inboundMessageRef.get(), Foo.class); - assertThat(fooPayload).isNotNull(); - assertThat(inboundMessageRef.get().getHeaders() - .get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)).isNull(); - producerBinding.unbind(); - consumerBinding.unbind(); - } - - @Test - @SuppressWarnings({ "rawtypes", "deprecation" }) - 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( - String.format("foo%s0y", getDestinationNameDelimiter()), - moduleOutputChannel, outputBindingProperties.getProducer()); - - Binding consumerBinding = binder.bindConsumer( - String.format("foo%s0y", getDestinationNameDelimiter()), - "testSendAndReceiveJavaSerialization", moduleInputChannel, - inputBindingProperties.getConsumer()); - 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(message1 -> { - try { - inboundMessageRef.set((Message) message1); - } - finally { - latch.countDown(); - } - }); - - moduleOutputChannel.send(message); - Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message"); - - JavaSerializationMessageConverter converter = new JavaSerializationMessageConverter(); - SerializableFoo serializableFoo = (SerializableFoo) converter.convertFromInternal( - inboundMessageRef.get(), SerializableFoo.class, null); - assertThat(serializableFoo).isNotNull(); - 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 { @@ -888,19 +778,4 @@ public abstract class AbstractBinderTestsspring-boot-starter-test test + + com.fasterxml.jackson.core + jackson-databind + test + 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 deleted file mode 100644 index 73be68443..000000000 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/InboundJsonToTupleConversionTest.java +++ /dev/null @@ -1,84 +0,0 @@ -/* - * Copyright 2017-2019 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://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.config; - -import java.nio.charset.StandardCharsets; -import java.util.concurrent.TimeUnit; - -import org.junit.Test; -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.EnableBinding; -import org.springframework.cloud.stream.binder.BinderFactory; -import org.springframework.cloud.stream.messaging.Processor; -import org.springframework.cloud.stream.test.binder.TestSupportBinder; -import org.springframework.context.annotation.PropertySource; -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.test.context.junit4.SpringJUnit4ClassRunner; -import org.springframework.tuple.Tuple; -import org.springframework.tuple.TupleBuilder; - -import static org.assertj.core.api.Assertions.assertThat; - -/** - * @author Marius Bogoevici - * @author Oleg Zhurakousky - */ -@RunWith(SpringJUnit4ClassRunner.class) -@SpringBootTest(classes = InboundJsonToTupleConversionTest.FooProcessor.class) -public class InboundJsonToTupleConversionTest { - - @Autowired - private Processor testProcessor; - - @Autowired - private BinderFactory binderFactory; - - @Test - public void testInboundJsonTupleConversion() throws Exception { - this.testProcessor.input() - .send(MessageBuilder.withPayload("{'name':'foo'}").build()); - @SuppressWarnings("unchecked") - Message received = (Message) ((TestSupportBinder) this.binderFactory - .getBinder(null, MessageChannel.class)).messageCollector() - .forChannel(this.testProcessor.output()) - .poll(1, TimeUnit.SECONDS); - assertThat(received).isNotNull(); - String payload = new String(received.getPayload(), StandardCharsets.UTF_8); - assertThat(TupleBuilder.fromString(payload)) - .isEqualTo(TupleBuilder.tuple().of("name", "foo")); - } - - @EnableBinding(Processor.class) - @EnableAutoConfiguration - @PropertySource("classpath:/org/springframework/cloud/stream/config/inboundjsontuple/inbound-json-tuple.properties") - public static class FooProcessor { - - @ServiceActivator(inputChannel = "input", outputChannel = "output") - public Tuple consume(Tuple tuple) { - return tuple; - } - - } - -} diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java index e6cadc049..55490339a 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java @@ -19,18 +19,14 @@ package org.springframework.cloud.stream.config; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; -import com.fasterxml.jackson.databind.ObjectMapper; -import com.fasterxml.jackson.databind.SerializationFeature; import org.junit.Test; import org.junit.runner.RunWith; -import org.springframework.beans.DirectFieldAccessor; 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.EnableBinding; import org.springframework.cloud.stream.binder.BinderHeaders; -import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel; import org.springframework.cloud.stream.messaging.Sink; import org.springframework.cloud.stream.messaging.Source; @@ -39,10 +35,7 @@ import org.springframework.context.annotation.PropertySource; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; -import org.springframework.messaging.converter.CompositeMessageConverter; -import org.springframework.messaging.converter.MessageConverter; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; -import org.springframework.util.MimeTypeUtils; import static org.assertj.core.api.Assertions.assertThat; @@ -63,9 +56,6 @@ public class MessageChannelConfigurerTests { @Autowired private Source testSource; - @Autowired - private CompositeMessageConverterFactory messageConverterFactory; - @Autowired private MessageCollector messageCollector; @@ -94,23 +84,6 @@ public class MessageChannelConfigurerTests { this.testSink.input().unsubscribe(messageHandler); } - @Test - public void testObjectMapperConfig() throws Exception { - CompositeMessageConverter converters = (CompositeMessageConverter) this.messageConverterFactory - .getMessageConverterForType(MimeTypeUtils.APPLICATION_JSON); - for (MessageConverter converter : converters.getConverters()) { - DirectFieldAccessor converterAccessor = new DirectFieldAccessor(converter); - ObjectMapper objectMapper = (ObjectMapper) converterAccessor - .getPropertyValue("objectMapper"); - // assert that the ObjectMapper used by the converters is compliant with the - // Boot configuration - assertThat(!objectMapper.getSerializationConfig().isEnabled( - SerializationFeature.WRITE_DATES_AS_TIMESTAMPS)).withFailMessage( - "SerializationFeature 'WRITE_DATES_AS_TIMESTAMPS' should be disabled"); - // assert that the globally set bean is used by the converters - } - } - @Test public void testPartitionHeader() throws Exception { this.testSource.output() 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 9a89f751c..5fd550122 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 2017-2018 the original author or authors. + * Copyright 2017-2019 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,16 +16,11 @@ package org.springframework.cloud.stream.config.contentType; -import java.io.ByteArrayOutputStream; -import java.io.ObjectOutputStream; import java.util.LinkedList; import java.util.Map; import java.util.concurrent.TimeUnit; -import com.esotericsoftware.kryo.Kryo; -import com.esotericsoftware.kryo.io.Output; import com.fasterxml.jackson.databind.ObjectMapper; -import org.junit.Ignore; import org.junit.Test; import org.springframework.boot.SpringApplication; @@ -33,8 +28,6 @@ import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.annotation.Input; import org.springframework.cloud.stream.annotation.StreamListener; -import org.springframework.cloud.stream.converter.KryoMessageConverter; -import org.springframework.cloud.stream.converter.MessageConverterUtils; import org.springframework.cloud.stream.messaging.Source; import org.springframework.cloud.stream.test.binder.MessageCollector; import org.springframework.context.ConfigurableApplicationContext; @@ -43,8 +36,6 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.handler.annotation.Headers; -import org.springframework.tuple.Tuple; -import org.springframework.tuple.TupleBuilder; import org.springframework.util.MimeType; import org.springframework.util.MimeTypeUtils; @@ -173,45 +164,6 @@ public class ContentTypeTests { } } - @Test - public void testSendJavaSerializable() throws Exception { - try (ConfigurableApplicationContext context = SpringApplication.run( - SourceApplication.class, "--server.port=0", "--spring.jmx.enabled=false", - "--spring.cloud.stream.bindings.output.contentType=application/x-java-serialized-object")) { - MessageCollector collector = context.getBean(MessageCollector.class); - Source source = context.getBean(Source.class); - User user = new User("Alice"); - source.output().send(MessageBuilder.withPayload(user).build()); - 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)); - User received = message.getPayload(); - assertThat(user.getName()).isEqualTo(received.getName()); - } - } - - @Test - public void testSendKryoSerialized() throws Exception { - try (ConfigurableApplicationContext context = SpringApplication.run( - SourceApplication.class, "--server.port=0", "--spring.jmx.enabled=false", - "--spring.cloud.stream.bindings.output.contentType=application/x-java-object")) { - MessageCollector collector = context.getBean(MessageCollector.class); - Source source = context.getBean(Source.class); - User user = new User("Alice"); - source.output().send(MessageBuilder.withPayload(user).build()); - Message message = (Message) collector.forChannel(source.output()) - .poll(1, TimeUnit.SECONDS); - 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()); - - } - } - @Test public void testSendStringType() throws Exception { try (ConfigurableApplicationContext context = SpringApplication.run( @@ -230,25 +182,6 @@ public class ContentTypeTests { } } - @Test - public void testSendTuple() throws Exception { - try (ConfigurableApplicationContext context = SpringApplication.run( - SourceApplication.class, "--server.port=0", "--spring.jmx.enabled=false", - "--spring.cloud.stream.bindings.output.contentType=application/x-spring-tuple")) { - MessageCollector collector = context.getBean(MessageCollector.class); - Source source = context.getBean(Source.class); - Tuple tuple = TupleBuilder.tuple().of("foo", "bar"); - source.output().send(MessageBuilder.withPayload(tuple).build()); - Message message = (Message) collector - .forChannel(source.output()).poll(1, TimeUnit.SECONDS); - assertThat( - message.getHeaders().get(MessageHeaders.CONTENT_TYPE, MimeType.class) - .includes(MessageConverterUtils.X_SPRING_TUPLE)); - assertThat(TupleBuilder.fromString(new String(message.getPayload()))) - .isEqualTo(tuple); - } - } - @Test public void testReceiveWithDefaults() throws Exception { try (ConfigurableApplicationContext context = SpringApplication.run( @@ -289,78 +222,6 @@ public class ContentTypeTests { } } - @Test - @Ignore - public void testReceiveKryoPayload() { - try (ConfigurableApplicationContext context = SpringApplication.run( - SinkApplication.class, "--server.port=0", "--debug", - "--spring.jmx.enabled=false", - "--spring.cloud.stream.bindings.pojo_input.contentType=" - + "application/x-java-object;type=org.springframework.cloud.stream.config.contentType.User")) { - TestSink testSink = context.getBean(TestSink.class); - SinkApplication sourceApp = context.getBean(SinkApplication.class); - Kryo kryo = new Kryo(); - User user = new User("Alice"); - ByteArrayOutputStream baos = new ByteArrayOutputStream(); - Output output = new Output(baos); - kryo.writeObject(output, user); - output.close(); - testSink.pojo().send(MessageBuilder.withPayload(baos.toByteArray()).build()); - Map headers = (Map) sourceApp.arguments.pop(); - User received = (User) sourceApp.arguments.pop(); - assertThat(((MimeType) headers.get(MessageHeaders.CONTENT_TYPE)) - .includes(MimeType.valueOf(KryoMessageConverter.KRYO_MIME_TYPE))); - assertThat(user.getName()).isEqualTo(received.getName()); - } - } - - @Test - @SuppressWarnings("deprecation") - public void testReceiveKryoWithHeadersOverridingDefault() { - try (ConfigurableApplicationContext context = SpringApplication.run( - SinkApplication.class, "--server.port=0", "--spring.jmx.enabled=false")) { - TestSink testSink = context.getBean(TestSink.class); - SinkApplication sourceApp = context.getBean(SinkApplication.class); - Kryo kryo = new Kryo(); - User user = new User("Alice"); - ByteArrayOutputStream baos = new ByteArrayOutputStream(); - Output output = new Output(baos); - kryo.writeObject(output, user); - output.close(); - testSink.pojo() - .send(MessageBuilder.withPayload(baos.toByteArray()) - .setHeader(MessageHeaders.CONTENT_TYPE, - MimeType.valueOf(KryoMessageConverter.KRYO_MIME_TYPE)) - .build()); - Map headers = (Map) sourceApp.arguments.pop(); - User received = (User) sourceApp.arguments.pop(); - assertThat(((MimeType) headers.get(MessageHeaders.CONTENT_TYPE)) - .includes(MimeType.valueOf(KryoMessageConverter.KRYO_MIME_TYPE))); - assertThat(user.getName()).isEqualTo(received.getName()); - } - } - - @Test - @Ignore - public void testReceiveJavaSerializable() throws Exception { - try (ConfigurableApplicationContext context = SpringApplication.run( - SinkApplication.class, "--server.port=0", "--spring.jmx.enabled=false", - "--spring.cloud.stream.bindings.pojo_input.contentType=application/x-java-serialized-object")) { - TestSink testSink = context.getBean(TestSink.class); - SinkApplication sourceApp = context.getBean(SinkApplication.class); - User user = new User("Alice"); - ByteArrayOutputStream baos = new ByteArrayOutputStream(); - new ObjectOutputStream(baos).writeObject(user); - testSink.pojo().send(MessageBuilder.withPayload(baos.toByteArray()).build()); - Map headers = (Map) sourceApp.arguments.pop(); - User received = (User) sourceApp.arguments.pop(); - assertThat(((MimeType) headers.get(MessageHeaders.CONTENT_TYPE)) - .includes(MessageConverterUtils.X_JAVA_SERIALIZED_OBJECT)); - assertThat(user.getName()).isEqualTo(received.getName()); - - } - } - public interface TestSink { @Input("POJO_INPUT") @@ -395,10 +256,6 @@ public class ContentTypeTests { this.arguments.push(headers); } - @StreamListener("TUPLE_INPUT") - public void receive(Tuple tuple) { - } - @StreamListener("STRING_INPUT") public void receive(String string) { } diff --git a/spring-cloud-stream-test-support/pom.xml b/spring-cloud-stream-test-support/pom.xml index a45710fe9..5ff343c50 100644 --- a/spring-cloud-stream-test-support/pom.xml +++ b/spring-cloud-stream-test-support/pom.xml @@ -29,5 +29,9 @@ org.springframework.boot spring-boot-autoconfigure + + com.fasterxml.jackson.core + jackson-databind + diff --git a/spring-cloud-stream/pom.xml b/spring-cloud-stream/pom.xml index 26addd01d..fc24029d4 100644 --- a/spring-cloud-stream/pom.xml +++ b/spring-cloud-stream/pom.xml @@ -37,14 +37,6 @@ org.springframework.integration spring-integration-jmx - - org.springframework - spring-tuple - - - org.springframework.integration - spring-integration-tuple - org.springframework.retry spring-retry diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/SharedChannelRegistry.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/SharedChannelRegistry.java deleted file mode 100644 index 1c22cdd49..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/SharedChannelRegistry.java +++ /dev/null @@ -1,67 +0,0 @@ -/* - * 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. - * You may obtain a copy of the License at - * - * https://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.aggregate; - -import java.util.Collections; -import java.util.HashMap; -import java.util.Map; - -import org.springframework.messaging.MessageChannel; - -/** - * Wraps the {@link SharedBindingTargetRegistry} for access to {@link MessageChannel} - * instances. This class is provided as a convenience for users of - * {@link SharedChannelRegistry} in previous versions and will be removed in the future. - * - * @author Marius Bogoevici - * @deprecated in favor of {@link SharedBindingTargetRegistry}. Will be removed in 3.0. - * Not currently used by the framework. - */ -@Deprecated -public class SharedChannelRegistry { - - private final SharedBindingTargetRegistry sharedBindingTargetRegistry; - - public SharedChannelRegistry( - SharedBindingTargetRegistry sharedBindingTargetRegistry) { - this.sharedBindingTargetRegistry = sharedBindingTargetRegistry; - } - - public MessageChannel get(String id) { - return this.sharedBindingTargetRegistry.get(id, MessageChannel.class); - } - - public void register(String id, MessageChannel bindingTarget) { - this.sharedBindingTargetRegistry.register(id, bindingTarget); - } - - public Map getAll() { - Map sharedBindingTargets = this.sharedBindingTargetRegistry - .getAll(); - Map sharedMessageChannels = new HashMap<>(); - for (Map.Entry sharedBindingTargetEntry : sharedBindingTargets - .entrySet()) { - if (MessageChannel.class - .isAssignableFrom(sharedBindingTargetEntry.getValue().getClass())) { - sharedMessageChannels.put(sharedBindingTargetEntry.getKey(), - (MessageChannel) sharedBindingTargetEntry.getValue()); - } - } - return Collections.unmodifiableMap(sharedMessageChannels); - } - -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Bindings.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Bindings.java deleted file mode 100644 index d710413ba..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Bindings.java +++ /dev/null @@ -1,48 +0,0 @@ -/* - * Copyright 2015-2019 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://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.annotation; - -import java.lang.annotation.Documented; -import java.lang.annotation.ElementType; -import java.lang.annotation.Inherited; -import java.lang.annotation.Retention; -import java.lang.annotation.RetentionPolicy; -import java.lang.annotation.Target; - -import org.springframework.beans.factory.annotation.Qualifier; - -/** - * Indicates an instance of an interface containing methods returning bound inputs and - * outputs. - * - * @author Dave Syer - * @author Marius Bogoevici - * @deprecated As of 1.1 for being redundant (beans qualified by it are already uniquely - * identified by their type) - */ - -@Qualifier -@Target({ ElementType.FIELD, ElementType.METHOD, ElementType.ANNOTATION_TYPE }) -@Retention(RetentionPolicy.RUNTIME) -@Inherited -@Documented -@Deprecated -public @interface Bindings { - - Class value(); - -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java index f5f979c63..2131daa3b 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2018 the original author or authors. + * Copyright 2015-2019 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -23,7 +23,7 @@ import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; -import org.springframework.cloud.stream.config.BinderFactoryConfiguration; +import org.springframework.cloud.stream.config.BinderFactoryAutoConfiguration; import org.springframework.cloud.stream.config.BindingBeansRegistrar; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; @@ -43,7 +43,7 @@ import org.springframework.integration.config.EnableIntegration; @Documented @Inherited @Configuration -@Import({ BindingBeansRegistrar.class, BinderFactoryConfiguration.class }) +@Import({ BindingBeansRegistrar.class, BinderFactoryAutoConfiguration.class }) @EnableIntegration public @interface EnableBinding { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java index 84ec35266..369ffc9db 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java @@ -32,7 +32,6 @@ import org.springframework.context.support.AbstractApplicationContext; import org.springframework.context.support.GenericApplicationContext; import org.springframework.expression.EvaluationContext; import org.springframework.integration.expression.ExpressionUtils; -import org.springframework.messaging.Message; import org.springframework.retry.backoff.ExponentialBackOffPolicy; import org.springframework.retry.policy.SimpleRetryPolicy; import org.springframework.retry.support.RetryTemplate; @@ -167,30 +166,6 @@ public abstract class AbstractBinder message) { - return new MessageValues(message); - } - - /** - * Deprecated as of v2.0. Remains primarily for backward compatibility and will be - * removed in the next major release. - * @param expressionRoot root of the expression - * @return full expression for a header - */ - @Deprecated - protected String buildPartitionRoutingExpression(String expressionRoot) { - return "'" + expressionRoot + "-' + headers['" + BinderHeaders.PARTITION_HEADER - + "']"; - } - /** * Create and configure a default retry template unless one has already been provided * via @Bean by an application. diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java index feb1cea3c..e399e73a6 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java @@ -137,14 +137,7 @@ public abstract class AbstractMessageChannelBinder containerCustomizer) { - - this(headersToEmbed, provisioningProvider, containerCustomizer, null); + this(headersToEmbed, provisioningProvider, null, null); } public AbstractMessageChannelBinder(String[] headersToEmbed, PP provisioningProvider, @@ -1106,10 +1099,9 @@ public abstract class AbstractMessageChannelBinder serializeAndEmbedHeadersIfApplicable(Message message) throws Exception { - MessageValues transformed = serializePayloadIfNecessary(message); + MessageValues transformed = new MessageValues(message); Object payload; if (this.embedHeaders) { Object contentType = transformed.get(MessageHeaders.CONTENT_TYPE); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/HeaderMode.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/HeaderMode.java index 96a120578..d88f5127e 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/HeaderMode.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/HeaderMode.java @@ -22,11 +22,6 @@ package org.springframework.cloud.stream.binder; */ public enum HeaderMode { - /** - * @deprecated - use {@link #none}. - */ - raw, - /** * No headers. */ diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java index de9971630..503dce137 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java @@ -18,7 +18,6 @@ package org.springframework.cloud.stream.binder; import java.io.IOException; -import javax.validation.constraints.AssertTrue; import javax.validation.constraints.Min; import com.fasterxml.jackson.annotation.JsonInclude; @@ -49,11 +48,11 @@ public class ProducerProperties { @JsonSerialize(using = ExpressionSerializer.class) private Expression partitionKeyExpression; - /** - * @deprecated in favor of 'partitionKeyExtractorName' - */ - @Deprecated - private Class partitionKeyExtractorClass; +// /** +// * @deprecated in favor of 'partitionKeyExtractorName' +// */ +// @Deprecated +// private Class partitionKeyExtractorClass; /** * The name of the bean that implements {@link PartitionKeyExtractorStrategy}\. Used @@ -62,11 +61,11 @@ public class ProducerProperties { */ private String partitionKeyExtractorName; - /** - * @deprecated in favor of 'partitionSelectorName' - */ - @Deprecated - private Class partitionSelectorClass; +// /** +// * @deprecated in favor of 'partitionSelectorName' +// */ +// @Deprecated +// private Class partitionSelectorClass; /** * The name of the bean that implements {@link PartitionSelectorStrategy}\. Used to @@ -96,31 +95,31 @@ public class ProducerProperties { this.partitionKeyExpression = partitionKeyExpression; } - @Deprecated - public Class getPartitionKeyExtractorClass() { - return this.partitionKeyExtractorClass; - } - - @Deprecated - public void setPartitionKeyExtractorClass(Class partitionKeyExtractorClass) { - this.partitionKeyExtractorClass = partitionKeyExtractorClass; - } +// @Deprecated +// public Class getPartitionKeyExtractorClass() { +// return this.partitionKeyExtractorClass; +// } +// +// @Deprecated +// public void setPartitionKeyExtractorClass(Class partitionKeyExtractorClass) { +// this.partitionKeyExtractorClass = partitionKeyExtractorClass; +// } public boolean isPartitioned() { return this.partitionKeyExpression != null - || this.partitionKeyExtractorName != null - || this.partitionKeyExtractorClass != null; + || this.partitionKeyExtractorName != null; +// || this.partitionKeyExtractorClass != null; } - @Deprecated - public Class getPartitionSelectorClass() { - return this.partitionSelectorClass; - } - - @Deprecated - public void setPartitionSelectorClass(Class partitionSelectorClass) { - this.partitionSelectorClass = partitionSelectorClass; - } +// @Deprecated +// public Class getPartitionSelectorClass() { +// return this.partitionSelectorClass; +// } +// +// @Deprecated +// public void setPartitionSelectorClass(Class partitionSelectorClass) { +// this.partitionSelectorClass = partitionSelectorClass; +// } public Expression getPartitionSelectorExpression() { return this.partitionSelectorExpression; @@ -147,16 +146,18 @@ public class ProducerProperties { this.requiredGroups = requiredGroups; } - @AssertTrue(message = "Partition key expression and partition key extractor class properties are mutually exclusive.") + //@AssertTrue(message = "Partition key expression and partition key extractor class properties are mutually exclusive.") public boolean isValidPartitionKeyProperty() { - return (this.partitionKeyExpression == null) - || (this.partitionKeyExtractorClass == null); +// return (this.partitionKeyExpression == null) +// || (this.partitionKeyExtractorClass == null); + return this.partitionKeyExpression == null; } - @AssertTrue(message = "Partition selector class and partition selector expression properties are mutually exclusive.") + //@AssertTrue(message = "Partition selector class and partition selector expression properties are mutually exclusive.") public boolean isValidPartitionSelectorProperty() { - return (this.partitionSelectorClass == null) - || (this.partitionSelectorExpression == null); +// return (this.partitionSelectorClass == null) +// || (this.partitionSelectorExpression == null); + return this.partitionSelectorExpression == null; } public HeaderMode getHeaderMode() { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/Bindable.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/Bindable.java index 07cb4b657..dcd3438db 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/Bindable.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/Bindable.java @@ -32,16 +32,6 @@ import org.springframework.cloud.stream.binder.Binding; */ public interface Bindable { - /** - * Binds all the inputs associated with this instance. - * @deprecated as of 2.0 in favor of {@link #createAndBindInputs(BindingService)} - * @param adapter binding service - */ - @Deprecated - default void bindInputs(BindingService adapter) { - this.createAndBindInputs(adapter); - } - /** * Binds all the inputs associated with this instance. * @param adapter instance of {@link BindingService} @@ -53,15 +43,6 @@ public interface Bindable { return Collections.>emptyList(); } - /** - * Binds all the outputs associated with this instance. - * @deprecated as of 2.0 in favor of {@link #createAndBindOutputs(BindingService)} - * @param adapter binding service - */ - @Deprecated - default void bindOutputs(BindingService adapter) { - } - /** * Binds all the outputs associated with this instance. * @param adapter instance of {@link BindingService} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java index b30b81337..43fe02b10 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2018 the original author or authors. + * Copyright 2015-2019 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -225,15 +225,6 @@ public class BindableProxyFactory return true; } - /** - * @deprecated in favor of {@link #createAndBindInputs(BindingService)} - */ - @Override - @Deprecated - public void bindInputs(BindingService bindingService) { - this.createAndBindInputs(bindingService); - } - @Override public Collection> createAndBindInputs( BindingService bindingService) { @@ -258,15 +249,6 @@ public class BindableProxyFactory return bindings; } - /** - * @deprecated in favor of {@link #createAndBindOutputs(BindingService)} - */ - @Override - @Deprecated - public void bindOutputs(BindingService bindingService) { - this.createAndBindOutputs(bindingService); - } - @Override public Collection> createAndBindOutputs( BindingService bindingService) { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java index f2b2c046b..95ec2a88a 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java @@ -21,7 +21,6 @@ import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.ProducerProperties; import org.springframework.cloud.stream.config.BindingServiceProperties; -import org.springframework.integration.config.GlobalChannelInterceptorProcessor; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.core.BeanFactoryMessageChannelDestinationResolver; import org.springframework.messaging.core.DestinationResolutionException; @@ -55,8 +54,7 @@ public class BinderAwareChannelResolver public BinderAwareChannelResolver(BindingService bindingService, AbstractBindingTargetFactory bindingTargetFactory, DynamicDestinationsBindable dynamicDestinationsBindable) { - this(bindingService, bindingTargetFactory, dynamicDestinationsBindable, null, - null); + this(bindingService, bindingTargetFactory, dynamicDestinationsBindable, null); } @SuppressWarnings("rawtypes") @@ -64,28 +62,6 @@ public class BinderAwareChannelResolver AbstractBindingTargetFactory bindingTargetFactory, DynamicDestinationsBindable dynamicDestinationsBindable, NewDestinationBindingCallback callback) { - this(bindingService, bindingTargetFactory, dynamicDestinationsBindable, callback, - null); - } - - /** - * @deprecated since GlobalChannelInterceptorProcessor is no longer used - * @param bindingService service to bind inputs and outputs - * @param bindingTargetFactory implementation that restricts the type of binding - * target to a specified class and its supertypes - * @param dynamicDestinationsBindable stores the dynamic destination names and handles - * their unbinding. - * @param callback used to configure a new destination before it is bound. - * @param globalChannelInterceptorProcessor applies global interceptors to message - * channel beans - */ - @SuppressWarnings("rawtypes") - @Deprecated - public BinderAwareChannelResolver(BindingService bindingService, - AbstractBindingTargetFactory bindingTargetFactory, - DynamicDestinationsBindable dynamicDestinationsBindable, - NewDestinationBindingCallback callback, - GlobalChannelInterceptorProcessor globalChannelInterceptorProcessor) { this.dynamicDestinationsBindable = dynamicDestinationsBindable; Assert.notNull(bindingService, "'bindingService' cannot be null"); Assert.notNull(bindingTargetFactory, "'bindingTargetFactory' cannot be null"); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareRouterBeanPostProcessor.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareRouter.java similarity index 81% rename from spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareRouterBeanPostProcessor.java rename to spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareRouter.java index e91a5e5e5..6d56f7dbd 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareRouterBeanPostProcessor.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareRouter.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2017 the original author or authors. + * Copyright 2013-2019 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -28,13 +28,10 @@ import org.springframework.messaging.core.DestinationResolver; * @author Mark Fisher * @author Gary Russell * @author Oleg Zhurakousky - * @deprecated as of 2.0, will be renamed/replaced as it is no longer a BPP and naming is - * a bit confusing */ -@Deprecated -public class BinderAwareRouterBeanPostProcessor { +public class BinderAwareRouter { - public BinderAwareRouterBeanPostProcessor(AbstractMappingMessageRouter[] routers, + public BinderAwareRouter(AbstractMappingMessageRouter[] routers, DestinationResolver channelResolver) { if (routers != null) { for (AbstractMappingMessageRouter router : routers) { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingBeanDefinitionRegistryUtils.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingBeanDefinitionRegistryUtils.java index 6d07f75a1..2f83b1b42 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingBeanDefinitionRegistryUtils.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingBeanDefinitionRegistryUtils.java @@ -24,7 +24,6 @@ import org.springframework.beans.factory.BeanDefinitionStoreException; import org.springframework.beans.factory.support.AutowireCandidateQualifier; import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.beans.factory.support.RootBeanDefinition; -import org.springframework.cloud.stream.annotation.Bindings; import org.springframework.cloud.stream.annotation.Input; import org.springframework.cloud.stream.annotation.Output; import org.springframework.core.annotation.AnnotationUtils; @@ -38,7 +37,6 @@ import org.springframework.util.StringUtils; * @author Dave Syer * @author Artem Bilan */ -@SuppressWarnings("deprecation") public abstract class BindingBeanDefinitionRegistryUtils { public static void registerInputBindingTargetBeanDefinition(String qualifierValue, @@ -101,16 +99,16 @@ public abstract class BindingBeanDefinitionRegistryUtils { if (type.isInterface()) { RootBeanDefinition rootBeanDefinition = new RootBeanDefinition( BindableProxyFactory.class); - rootBeanDefinition - .addQualifier(new AutowireCandidateQualifier(Bindings.class, parent)); +// rootBeanDefinition +// .addQualifier(new AutowireCandidateQualifier(Bindings.class, parent)); rootBeanDefinition.getConstructorArgumentValues() .addGenericArgumentValue(type); registry.registerBeanDefinition(type.getName(), rootBeanDefinition); } else { RootBeanDefinition rootBeanDefinition = new RootBeanDefinition(type); - rootBeanDefinition - .addQualifier(new AutowireCandidateQualifier(Bindings.class, parent)); +// rootBeanDefinition +// .addQualifier(new AutowireCandidateQualifier(Bindings.class, parent)); registry.registerBeanDefinition(type.getName(), rootBeanDefinition); } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java index 554c675b8..2cfa8386c 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java @@ -318,15 +318,6 @@ public class BindingService { } } - /** - * Provided for backwards compatibility. Will be removed in a future version. - * @return {@link BindingServiceProperties} - */ - @Deprecated - public BindingServiceProperties getChannelBindingServiceProperties() { - return this.bindingServiceProperties; - } - public BindingServiceProperties getBindingServiceProperties() { return this.bindingServiceProperties; } 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 38cf7d7ce..f5ff7bda2 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 @@ -1,5 +1,5 @@ /* - * Copyright 2015-2018 the original author or authors. + * Copyright 2015-2019 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -19,14 +19,10 @@ package org.springframework.cloud.stream.binding; import java.lang.reflect.Field; import java.util.Map; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; -import org.springframework.cloud.stream.binder.BinderException; import org.springframework.cloud.stream.binder.BinderHeaders; import org.springframework.cloud.stream.binder.ConsumerProperties; import org.springframework.cloud.stream.binder.DefaultPollableMessageSource; @@ -76,8 +72,6 @@ import org.springframework.util.StringUtils; public class MessageConverterConfigurer implements MessageChannelAndSourceConfigurer, BeanFactoryAware { - private final Log logger = LogFactory.getLog(getClass()); - private final MessageBuilderFactory messageBuilderFactory = new MutableMessageBuilderFactory(); private final CompositeMessageConverterFactory compositeMessageConverterFactory; @@ -177,21 +171,10 @@ public class MessageConverterConfigurer } } - @SuppressWarnings("deprecation") private PartitionKeyExtractorStrategy getPartitionKeyExtractorStrategy( ProducerProperties producerProperties) { PartitionKeyExtractorStrategy partitionKeyExtractor; - if (producerProperties.getPartitionKeyExtractorClass() != null) { - this.logger.warn( - "'partitionKeyExtractorClass' option is deprecated as of v2.0. Please configure partition " - + "key extractor as a @Bean that implements 'PartitionKeyExtractorStrategy'. Additionally you can " - + "specify 'spring.cloud.stream.bindings.output.producer.partitionKeyExtractorName' to specify which " - + "bean to use in the event there are more then one."); - partitionKeyExtractor = instantiate( - producerProperties.getPartitionKeyExtractorClass(), - PartitionKeyExtractorStrategy.class); - } - else if (StringUtils.hasText(producerProperties.getPartitionKeyExtractorName())) { + if (StringUtils.hasText(producerProperties.getPartitionKeyExtractorName())) { partitionKeyExtractor = this.beanFactory.getBean( producerProperties.getPartitionKeyExtractorName(), PartitionKeyExtractorStrategy.class); @@ -214,21 +197,10 @@ public class MessageConverterConfigurer return partitionKeyExtractor; } - @SuppressWarnings("deprecation") private PartitionSelectorStrategy getPartitionSelectorStrategy( ProducerProperties producerProperties) { PartitionSelectorStrategy partitionSelector; - if (producerProperties.getPartitionSelectorClass() != null) { - this.logger.warn( - "'partitionSelectorClass' option is deprecated as of v2.0. Please configure partition " - + "selector as a @Bean that implements 'PartitionSelectorStrategy'. Additionally you can " - + "specify 'spring.cloud.stream.bindings.output.producer.partitionSelectorName' to specify which " - + "bean to use in the event there are more then one."); - partitionSelector = instantiate( - producerProperties.getPartitionSelectorClass(), - PartitionSelectorStrategy.class); - } - else if (StringUtils.hasText(producerProperties.getPartitionSelectorName())) { + if (StringUtils.hasText(producerProperties.getPartitionSelectorName())) { partitionSelector = this.beanFactory.getBean( producerProperties.getPartitionSelectorName(), PartitionSelectorStrategy.class); @@ -252,17 +224,6 @@ public class MessageConverterConfigurer return partitionSelector; } - @SuppressWarnings("unchecked") - private T instantiate(Class implClass, Class type) { - try { - return (T) implClass.newInstance(); - } - catch (Exception e) { - throw new BinderException( - "Failed to instantiate class: " + implClass.getName(), e); - } - } - /** * Default partition strategy; only works on keys with "real" hash codes, such as * String. Caller now always applies modulo so no need to do so here. diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SingleBindingTargetBindable.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SingleBindingTargetBindable.java deleted file mode 100644 index d66e59c5e..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SingleBindingTargetBindable.java +++ /dev/null @@ -1,72 +0,0 @@ -/* - * Copyright 2016-2019 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://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.binding; - -import java.util.Arrays; -import java.util.Collection; -import java.util.Collections; -import java.util.HashSet; -import java.util.Set; - -import org.springframework.cloud.stream.binder.Binding; - -/** - * A {@link Bindable} component that wraps a generic output binding target. Useful for - * binding targets outside the {@link org.springframework.cloud.stream.annotation.Input} - * and {@link org.springframework.cloud.stream.annotation.Output} annotated interfaces. - * - * @param type of binding target - * @author Ilayaperumal Gopinathan - * @author Marius Bogoevici - * @deprecated This class is no longer used by the framework and maybe removed in a future - * release. - */ -@Deprecated -public class SingleBindingTargetBindable implements Bindable { - - private final String name; - - private final T bindingTarget; - - public SingleBindingTargetBindable(String name, T bindingTarget) { - this.name = name; - this.bindingTarget = bindingTarget; - } - - @Override - public void bindOutputs(BindingService bindingService) { - this.createAndBindOutputs(bindingService); - } - - @Override - public Collection> createAndBindOutputs( - BindingService bindingService) { - return Collections.singletonList( - bindingService.bindProducer(this.bindingTarget, this.name)); - } - - @Override - public void unbindOutputs(BindingService bindingService) { - bindingService.unbindProducers(this.name); - } - - @Override - public Set getOutputs() { - return Collections.unmodifiableSet(new HashSet<>(Arrays.asList(this.name))); - } - -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java similarity index 98% rename from spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryConfiguration.java rename to spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java index c32f680ca..a33f9b61a 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java @@ -90,15 +90,12 @@ import org.springframework.validation.Validator; * @author Oleg Zhurakousky * @author Soby Chacko * @author David Harrigan - * @deprecated since it really represents 'auto-configuration' it will be - * renamed/restructured in the next release. */ @Configuration @Role(BeanDefinition.ROLE_INFRASTRUCTURE) @EnableConfigurationProperties({ BindingServiceProperties.class }) @Import(ContentTypeConfiguration.class) -@Deprecated -public class BinderFactoryConfiguration { +public class BinderFactoryAutoConfiguration { private static final String SPRING_CLOUD_STREAM_INTERNAL_PREFIX = "spring.cloud.stream.internal"; diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderProperties.java index 944458894..30f63b769 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderProperties.java @@ -18,8 +18,6 @@ package org.springframework.cloud.stream.config; import java.util.HashMap; import java.util.Map; -import java.util.Properties; -import java.util.stream.Collectors; /** * Contains the properties of a binder. @@ -66,16 +64,6 @@ public class BinderProperties { return this.environment; } - /** - * @deprecated in 2.0.0 in preference to {@link #setEnvironment(Map)} - * @param environment properties to which stream props will be added - */ - @Deprecated - public void setEnvironment(Properties environment) { - this.environment.clear(); - this.environment.putAll(environment.entrySet().stream().collect( - Collectors.toMap(e -> e.getKey().toString(), e -> e.getValue()))); - } public void setEnvironment(Map environment) { this.environment = environment; diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java index b7796ac16..47a7fee0f 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java @@ -39,6 +39,7 @@ import org.springframework.cloud.stream.binder.DefaultBinderFactory; import org.springframework.cloud.stream.binding.AbstractBindingTargetFactory; import org.springframework.cloud.stream.binding.Bindable; import org.springframework.cloud.stream.binding.BinderAwareChannelResolver; +import org.springframework.cloud.stream.binding.BinderAwareRouter; import org.springframework.cloud.stream.binding.BindingService; import org.springframework.cloud.stream.binding.ContextStartAfterRefreshListener; import org.springframework.cloud.stream.binding.DynamicDestinationsBindable; @@ -242,16 +243,14 @@ public class BindingServiceConfiguration { return new DynamicDestinationsBindable(); } - @SuppressWarnings("deprecation") @Bean @ConditionalOnMissingBean - public org.springframework.cloud.stream.binding.BinderAwareRouterBeanPostProcessor binderAwareRouterBeanPostProcessor( + public BinderAwareRouter binderAwareRouterBeanPostProcessor( @Autowired(required = false) AbstractMappingMessageRouter[] routers, @Autowired(required = false) @Qualifier("binderAwareChannelResolver") DestinationResolver channelResolver) { - return new org.springframework.cloud.stream.binding.BinderAwareRouterBeanPostProcessor( - routers, channelResolver); + return new BinderAwareRouter(routers, channelResolver); } @Bean diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/SmartMessageMethodArgumentResolver.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/SmartMessageMethodArgumentResolver.java index a46e7906b..6df0f6a25 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/SmartMessageMethodArgumentResolver.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/SmartMessageMethodArgumentResolver.java @@ -34,9 +34,7 @@ import org.springframework.util.StringUtils; /** * @author Oleg Zhurakousky - * @deprecated will be removed once https://jira.spring.io/browse/SPR-17503 is addressed */ -@Deprecated class SmartMessageMethodArgumentResolver extends MessageMethodArgumentResolver { private final MessageConverter messageConverter; diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/SmartPayloadArgumentResolver.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/SmartPayloadArgumentResolver.java index 839caf116..d075006b6 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/SmartPayloadArgumentResolver.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/SmartPayloadArgumentResolver.java @@ -38,10 +38,7 @@ import org.springframework.validation.Validator; /** * @author Oleg Zhurakousky * @author Gary Russell - * @deprecated will be removed once https://jira.spring.io/browse/SPR-17503 is addressed - * (but see note about KafkaNull below). */ -@Deprecated class SmartPayloadArgumentResolver extends PayloadArgumentResolver { private final MessageConverter messageConverter; diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/SpelExpressionConverterConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/SpelExpressionConverterConfiguration.java index 66acbfa9d..81c5fd9c6 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/SpelExpressionConverterConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/SpelExpressionConverterConfiguration.java @@ -36,7 +36,6 @@ import org.springframework.integration.config.IntegrationConverter; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.expression.SpelPropertyAccessorRegistrar; import org.springframework.integration.json.JsonPropertyAccessor; -import org.springframework.tuple.spel.TuplePropertyAccessor; /** * Adds a Converter from String to SpEL Expression in the context. @@ -62,10 +61,7 @@ public class SpelExpressionConverterConfiguration { return new SpelPropertyAccessorRegistrar() .add(Introspector .decapitalize(JsonPropertyAccessor.class.getSimpleName()), - new JsonPropertyAccessor()) - .add(Introspector - .decapitalize(TuplePropertyAccessor.class.getSimpleName()), - new TuplePropertyAccessor()); + new JsonPropertyAccessor()); } @Bean 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 147152ab5..7bc77af97 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 @@ -78,13 +78,11 @@ public class CompositeMessageConverterFactory { .setContentTypeResolver(resolver)); } - @SuppressWarnings("deprecation") private void initDefaultConverters() { ApplicationJsonMessageMarshallingConverter applicationJsonConverter = new ApplicationJsonMessageMarshallingConverter( this.objectMapper); applicationJsonConverter.setStrictContentTypeMatch(true); this.converters.add(applicationJsonConverter); - this.converters.add(new TupleJsonMessageConverter(this.objectMapper)); this.converters.add(new ByteArrayMessageConverter() { @Override protected boolean supports(Class clazz) { @@ -95,11 +93,6 @@ public class CompositeMessageConverterFactory { } }); this.converters.add(new ObjectStringMessageConverter()); - - // Deprecated converters - this.converters.add(new JavaSerializationMessageConverter()); - this.converters.add(new KryoMessageConverter(null, true)); - this.converters.add(new JsonUnmarshallingConverter(this.objectMapper)); } /** 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 deleted file mode 100644 index 4d0e8c529..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JavaSerializationMessageConverter.java +++ /dev/null @@ -1,82 +0,0 @@ -/* - * Copyright 2016-2018 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 - * - * https://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.io.ByteArrayInputStream; -import java.io.ByteArrayOutputStream; -import java.io.IOException; -import java.io.ObjectInputStream; -import java.io.ObjectOutputStream; -import java.io.Serializable; -import java.util.Arrays; - -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHeaders; -import org.springframework.messaging.converter.AbstractMessageConverter; - -/** - * @author Marius Bogoevici - * @author Oleg Zhurakousky - * @deprecated as of 2.0. Will be removed in 2.1 - */ -@Deprecated -public class JavaSerializationMessageConverter extends AbstractMessageConverter { - - public JavaSerializationMessageConverter() { - super(Arrays.asList(MessageConverterUtils.X_JAVA_SERIALIZED_OBJECT)); - } - - @Override - protected boolean supports(Class clazz) { - if (clazz != null) { - return Serializable.class.isAssignableFrom(clazz); - } - return true; - } - - @Override - public Object convertFromInternal(Message message, Class targetClass, - Object conversionHint) { - if (!(message.getPayload() instanceof byte[])) { - return null; - } - try { - ByteArrayInputStream bis = new ByteArrayInputStream( - (byte[]) (message.getPayload())); - return new ObjectInputStream(bis).readObject(); - } - catch (Exception e) { - this.logger.error(e.getMessage(), e); - } - return null; - } - - @Override - protected Object convertToInternal(Object payload, MessageHeaders headers, - Object conversionHint) { - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - try { - new ObjectOutputStream(bos).writeObject(payload); - } - catch (IOException e) { - this.logger.error(e.getMessage(), e); - return null; - } - return bos.toByteArray(); - } - -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonUnmarshallingConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonUnmarshallingConverter.java deleted file mode 100644 index 35450c1b4..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonUnmarshallingConverter.java +++ /dev/null @@ -1,74 +0,0 @@ -/* - * Copyright 2016-2019 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://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.io.IOException; - -import com.fasterxml.jackson.databind.ObjectMapper; - -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHeaders; -import org.springframework.messaging.converter.AbstractMessageConverter; -import org.springframework.messaging.converter.MessageConversionException; - -/** - * Message converter providing backwards compatibility for applications using an Java type - * as input. - * - * @author Marius Bogoevici - * @deprecated as of 2.0. - */ -// NOTE we need to revisit as to why do we need it in the first place, given that our -// first converter already handles JSON -@Deprecated -public class JsonUnmarshallingConverter extends AbstractMessageConverter { - - private final ObjectMapper objectMapper; - - protected JsonUnmarshallingConverter(ObjectMapper objectMapper) { - super(MessageConverterUtils.X_JAVA_OBJECT); - setStrictContentTypeMatch(true); - this.objectMapper = objectMapper != null ? objectMapper : new ObjectMapper(); - } - - @Override - protected boolean supports(Class aClass) { - return String.class.isAssignableFrom(aClass) - || byte[].class.isAssignableFrom(aClass); - } - - @Override - protected Object convertFromInternal(Message message, Class targetClass, - Object conversionHint) { - Object payload = message.getPayload(); - try { - return payload instanceof byte[] - ? this.objectMapper.readValue((byte[]) payload, targetClass) - : this.objectMapper.readValue((String) payload, targetClass); - } - catch (IOException e) { - throw new MessageConversionException("Cannot parse payload ", e); - } - } - - @Override - protected Object convertToInternal(Object payload, MessageHeaders headers, - Object conversionHint) { - return super.convertToInternal(payload, headers, conversionHint); - } - -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/KryoMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/KryoMessageConverter.java deleted file mode 100644 index b143f6fcb..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/KryoMessageConverter.java +++ /dev/null @@ -1,265 +0,0 @@ -/* - * Copyright 2017-2019 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://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.io.ByteArrayOutputStream; -import java.io.IOException; -import java.io.InputStream; -import java.util.Collections; -import java.util.List; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; - -import com.esotericsoftware.kryo.Kryo; -import com.esotericsoftware.kryo.io.Input; -import com.esotericsoftware.kryo.io.Output; -import com.esotericsoftware.kryo.pool.KryoFactory; -import com.esotericsoftware.kryo.pool.KryoPool; -import org.objenesis.strategy.StdInstantiatorStrategy; - -import org.springframework.integration.codec.kryo.CompositeKryoRegistrar; -import org.springframework.integration.codec.kryo.KryoRegistrar; -import org.springframework.lang.Nullable; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHeaders; -import org.springframework.messaging.converter.ContentTypeResolver; -import org.springframework.messaging.converter.DefaultContentTypeResolver; -import org.springframework.messaging.converter.MessageConversionException; -import org.springframework.messaging.converter.SmartMessageConverter; -import org.springframework.messaging.support.MessageBuilder; -import org.springframework.messaging.support.MessageHeaderAccessor; -import org.springframework.util.Assert; -import org.springframework.util.CollectionUtils; -import org.springframework.util.MimeType; - -/** - * @author Vinicius Carvalho - * @deprecated as of 2.0 all language specific type converters (kryo, java etc) are - * deprecated and won't be supported in the future. - */ -@Deprecated -public class KryoMessageConverter implements SmartMessageConverter { - - /** - * Kryo mime type. - */ - public static final String KRYO_MIME_TYPE = "application/x-java-object"; - - protected final KryoPool pool; - - private final CompositeKryoRegistrar kryoRegistrar; - - private final boolean useReferences; - - private final List supportedMimeTypes; - - private ConcurrentMap mimeTypesCache = new ConcurrentHashMap<>(); - - private ContentTypeResolver contentTypeResolver = new DefaultContentTypeResolver(); - - public KryoMessageConverter(List kryoRegistrars, - boolean useReferences) { - this.useReferences = useReferences; - this.kryoRegistrar = CollectionUtils.isEmpty(kryoRegistrars) ? null - : new CompositeKryoRegistrar(kryoRegistrars); - KryoFactory factory = () -> { - Kryo kryo = new Kryo(); - configureKryoInstance(kryo); - return kryo; - }; - this.pool = new KryoPool.Builder(factory).softReferences().build(); - this.supportedMimeTypes = Collections - .singletonList(MimeType.valueOf(KRYO_MIME_TYPE)); - } - - @Nullable - @Override - public Object fromMessage(Message message, Class targetClass, - @Nullable Object conversionHint) { - if (!canConvertFrom(message, targetClass)) { - return null; - } - if (!message.getPayload().getClass().isAssignableFrom(byte[].class)) { - throw new MessageConversionException( - "This converter can only convert messages with byte[] payload"); - } - byte[] payload = (byte[]) message.getPayload(); - try { - return deserialize(payload, targetClass); - } - catch (IOException e) { - throw new MessageConversionException("Could not deserialize payload", e); - } - } - - @Nullable - @Override - public Message toMessage(Object payload, @Nullable MessageHeaders headers, - @Nullable Object conversionHint) { - if (!canConvertTo(payload, headers)) { - return null; - } - byte[] payloadToUse = serialize(payload); - MimeType mimeType = getDefaultContentType(payload); - if (headers != null) { - MessageHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(headers, - MessageHeaderAccessor.class); - if (accessor != null && accessor.isMutable()) { - if (mimeType != null) { - accessor.setHeader(MessageHeaders.CONTENT_TYPE, mimeType); - } - return MessageBuilder.createMessage(payloadToUse, - accessor.getMessageHeaders()); - } - } - MessageBuilder builder = MessageBuilder.withPayload(payloadToUse); - if (headers != null) { - builder.copyHeaders(headers); - } - if (mimeType != null) { - builder.setHeader(MessageHeaders.CONTENT_TYPE, mimeType); - } - return builder.build(); - } - - private boolean canConvertTo(Object payload, MessageHeaders headers) { - return (supports(payload.getClass()) && supportsMimeType(headers)); - } - - @Nullable - protected MimeType getDefaultContentType(Object payload) { - return mimeTypeFromObject(payload); - } - - protected byte[] serialize(Object payload) { - ByteArrayOutputStream baos = new ByteArrayOutputStream(); - final Output output = new Output(baos); - this.pool.run(kryo -> { - kryo.writeObject(output, payload); - return Void.TYPE; - }); - output.close(); - return baos.toByteArray(); - } - - protected T deserialize(byte[] bytes, Class type) throws IOException { - Assert.notNull(bytes, "'bytes' cannot be null"); - final Input input = new Input(bytes); - try { - return deserialize(input, type); - } - finally { - input.close(); - } - } - - protected T deserialize(InputStream inputStream, final Class type) - throws IOException { - Assert.notNull(inputStream, "'inputStream' cannot be null"); - Assert.notNull(type, "'type' cannot be null"); - final Input input = (inputStream instanceof Input ? (Input) inputStream - : new Input(inputStream)); - T result = null; - try { - result = this.pool.run(kryo -> kryo.readObject(input, type)); - } - finally { - input.close(); - } - return result; - } - - protected void configureKryoInstance(Kryo kryo) { - kryo.setInstantiatorStrategy( - new Kryo.DefaultInstantiatorStrategy(new StdInstantiatorStrategy())); - if (this.kryoRegistrar != null) { - this.kryoRegistrar.registerTypes(kryo); - } - kryo.setReferences(this.useReferences); - } - - protected MimeType mimeTypeFromObject(Object payload) { - Assert.notNull(payload, "payload object cannot be null."); - String className = payload.getClass().getName(); - MimeType mimeType = this.mimeTypesCache.get(className); - if (mimeType == null) { - String modifiedClassName = className; - if (payload.getClass().isArray()) { - // Need to remove trailing ';' for an object array, e.g. - // "[Ljava.lang.String;" or multi-dimensional - // "[[[Ljava.lang.String;" - if (modifiedClassName.endsWith(";")) { - modifiedClassName = modifiedClassName.substring(0, - modifiedClassName.length() - 1); - } - // Wrap in quotes to handle the illegal '[' character - modifiedClassName = "\"" + modifiedClassName + "\""; - } - mimeType = MimeType.valueOf(KRYO_MIME_TYPE + ";type=" + modifiedClassName); - this.mimeTypesCache.put(className, mimeType); - } - return mimeType; - } - - protected boolean canConvertFrom(Message message, Class targetClass) { - return (supports(targetClass) && supportsMimeType(message.getHeaders())); - } - - protected boolean supportsMimeType(@Nullable MessageHeaders headers) { - if (getSupportedMimeTypes().isEmpty()) { - return true; - } - MimeType mimeType = getMimeType(headers); - if (mimeType == null) { - return false; - } - for (MimeType current : getSupportedMimeTypes()) { - if (current.getType().equals(mimeType.getType()) - && current.getSubtype().equals(mimeType.getSubtype())) { - return true; - } - } - return false; - } - - @Nullable - protected MimeType getMimeType(@Nullable MessageHeaders headers) { - return (headers != null && this.contentTypeResolver != null - ? this.contentTypeResolver.resolve(headers) : null); - } - - private boolean supports(Class targetClass) { - return true; - } - - @Nullable - @Override - public Object fromMessage(Message message, Class targetClass) { - return fromMessage(message, targetClass, null); - } - - @Nullable - @Override - public Message toMessage(Object payload, @Nullable MessageHeaders headers) { - return toMessage(payload, headers, null); - } - - public List getSupportedMimeTypes() { - return this.supportedMimeTypes; - } - -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/MessageConverterUtils.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/MessageConverterUtils.java index f39f2ccb0..4422e1455 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/MessageConverterUtils.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/MessageConverterUtils.java @@ -16,7 +16,6 @@ package org.springframework.cloud.stream.converter; -import org.springframework.tuple.Tuple; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; import org.springframework.util.MimeType; diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/TupleJsonMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/TupleJsonMessageConverter.java deleted file mode 100644 index 35a6417d2..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/TupleJsonMessageConverter.java +++ /dev/null @@ -1,101 +0,0 @@ -/* - * 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. - * You may obtain a copy of the License at - * - * https://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.io.IOException; -import java.nio.charset.Charset; -import java.util.Arrays; - -import com.fasterxml.jackson.databind.ObjectMapper; - -import org.springframework.beans.factory.annotation.Value; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHeaders; -import org.springframework.messaging.converter.AbstractMessageConverter; -import org.springframework.tuple.Tuple; -import org.springframework.tuple.TupleBuilder; -import org.springframework.util.MimeTypeUtils; - -/** - * A {@link org.springframework.messaging.converter.MessageConverter} to convert a - * {@link Tuple} to JSON bytes. - * - * @author David Turanski - * @author Ilayaperumal Gopinathan - * @author Marius Bogoevici - * @author Vinicius Carvalho - * @deprecated as of 2.0. please use 'application/json' content type - */ -@Deprecated -public class TupleJsonMessageConverter extends AbstractMessageConverter { - - private final ObjectMapper objectMapper; - - @Value("${typeconversion.json.prettyPrint:false}") - private volatile boolean prettyPrint; - - public TupleJsonMessageConverter(ObjectMapper objectMapper) { - super(Arrays.asList(MessageConverterUtils.X_SPRING_TUPLE, - MimeTypeUtils.APPLICATION_JSON)); - this.objectMapper = (objectMapper != null) ? objectMapper : new ObjectMapper(); - } - - public void setPrettyPrint(boolean prettyPrint) { - this.prettyPrint = prettyPrint; - } - - @Override - protected boolean supports(Class clazz) { - return Tuple.class.isAssignableFrom(clazz); - } - - @Override - protected Object convertToInternal(Object payload, MessageHeaders headers, - Object conversionHint) { - Tuple t = (Tuple) payload; - String json; - if (this.prettyPrint) { - try { - Object tmp = this.objectMapper.readValue(t.toString(), Object.class); - json = this.objectMapper.writerWithDefaultPrettyPrinter() - .writeValueAsString(tmp); - } - catch (IOException e) { - this.logger.error(e.getMessage(), e); - return null; - } - } - else { - json = t.toString(); - } - return json.getBytes(); - } - - @Override - public Object convertFromInternal(Message message, Class targetClass, - Object conversionHint) { - String source; - if (message.getPayload() instanceof byte[]) { - source = new String((byte[]) message.getPayload(), Charset.forName("UTF-8")); - } - else { - source = message.getPayload().toString(); - } - return TupleBuilder.fromString(source); - } - -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java index e8606f8fe..5a69dc55c 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java @@ -16,7 +16,6 @@ package org.springframework.cloud.stream.function; -import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; @@ -26,7 +25,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.function.context.FunctionCatalog; import org.springframework.cloud.function.context.catalog.FunctionInspector; -import org.springframework.cloud.stream.config.BinderFactoryConfiguration; +import org.springframework.cloud.stream.config.BinderFactoryAutoConfiguration; import org.springframework.cloud.stream.config.BindingServiceConfiguration; import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; @@ -50,7 +49,7 @@ import org.springframework.messaging.SubscribableChannel; */ @Configuration @EnableConfigurationProperties(StreamFunctionProperties.class) -@Import(BinderFactoryConfiguration.class) +@Import(BinderFactoryAutoConfiguration.class) @AutoConfigureBefore(BindingServiceConfiguration.class) public class FunctionConfiguration { @@ -89,13 +88,7 @@ public class FunctionConfiguration { public IntegrationFlow integrationFlowCreator( IntegrationFlowFunctionSupport functionSupport, @Nullable Source source, @Nullable Processor processor, @Nullable Sink sink) { - if (functionSupport.containsFunction(Consumer.class) - && consumerBindingPresent(processor, sink)) { - return functionSupport - .integrationFlowForFunction(getInputChannel(processor, sink), getOutputChannel(processor, source)) - .get(); - } - else if (functionSupport.containsFunction(Function.class) + if (functionSupport.containsFunction(Function.class) && consumerBindingPresent(processor, sink)) { return functionSupport .integrationFlowForFunction(getInputChannel(processor, sink), getOutputChannel(processor, source)) diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryConfigurationTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryAutoConfigurationTests.java similarity index 97% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryConfigurationTests.java rename to spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryAutoConfigurationTests.java index 77257c50b..c398e6766 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryConfigurationTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryAutoConfigurationTests.java @@ -31,7 +31,7 @@ import org.springframework.cloud.stream.binder.stub1.StubBinder1Configuration; import org.springframework.cloud.stream.binder.stub2.StubBinder2; import org.springframework.cloud.stream.binder.stub2.StubBinder2ConfigurationA; import org.springframework.cloud.stream.binder.stub2.StubBinder2ConfigurationB; -import org.springframework.cloud.stream.config.BinderFactoryConfiguration; +import org.springframework.cloud.stream.config.BinderFactoryAutoConfiguration; import org.springframework.cloud.stream.config.BindingServiceConfiguration; import org.springframework.context.ApplicationContext; import org.springframework.context.ConfigurableApplicationContext; @@ -52,7 +52,7 @@ import static org.junit.Assert.fail; * @author Soby Chacko * @author Artem Bilan */ -public class BinderFactoryConfigurationTests { +public class BinderFactoryAutoConfigurationTests { private static ClassLoader createClassLoader(String[] additionalClasspathDirectories, String... properties) throws IOException { @@ -65,7 +65,7 @@ public class BinderFactoryConfigurationTests { } } return new URLClassLoader(urls, - BinderFactoryConfigurationTests.class.getClassLoader()); + BinderFactoryAutoConfigurationTests.class.getClassLoader()); } private static ConfigurableApplicationContext createBinderTestContext( @@ -136,8 +136,7 @@ public class BinderFactoryConfigurationTests { ConfigurableApplicationContext context = createBinderTestContext( new String[] { "binder1" }, "binder1.name=foo", "spring.cloud.stream.binders.custom.environment.foo=bar", - "spring.cloud.stream.binders.custom.environment.spring.main.sources=" - + "org.springframework.cloud.stream.binder.BinderFactoryConfigurationTests.AdditionalBinderConfiguration", + "spring.cloud.stream.binders.custom.environment.spring.main.sources=" + AdditionalBinderConfiguration.class.getName(), "spring.cloud.stream.binders.custom.type=binder1"); BinderFactory binderFactory = context.getBean(BinderFactory.class); @@ -275,7 +274,7 @@ public class BinderFactoryConfigurationTests { assertThat(defaultBinder).isSameAs(binder2); } - @Import({ BinderFactoryConfiguration.class, + @Import({ BinderFactoryAutoConfiguration.class, PropertyPlaceholderAutoConfiguration.class, BindingServiceConfiguration.class }) @EnableBinding diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/HealthIndicatorsConfigurationTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/HealthIndicatorsConfigurationTests.java index acaed2129..9683bbf54 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/HealthIndicatorsConfigurationTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/HealthIndicatorsConfigurationTests.java @@ -65,7 +65,7 @@ public class HealthIndicatorsConfigurationTests { } } ClassLoader classLoader = new URLClassLoader(urls, - BinderFactoryConfigurationTests.class.getClassLoader()); + BinderFactoryAutoConfigurationTests.class.getClassLoader()); return new SpringApplicationBuilder(SimpleSource.class) .resourceLoader(new DefaultResourceLoader(classLoader)) diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java index 9efd50cba..993d69d89 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java @@ -34,8 +34,6 @@ import org.springframework.cloud.stream.binder.test.InputDestination; import org.springframework.cloud.stream.binder.test.OutputDestination; import org.springframework.cloud.stream.binder.test.TestChannelBinder; import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; -import org.springframework.cloud.stream.converter.KryoMessageConverter; -import org.springframework.cloud.stream.converter.MessageConverterUtils; import org.springframework.cloud.stream.messaging.Processor; import org.springframework.context.ApplicationContext; import org.springframework.context.annotation.Bean; @@ -493,84 +491,6 @@ public class ContentTypeTckTests { .isEqualTo("oleg"); } - @Test - @SuppressWarnings("deprecation") - public void kryo_pojoToPojo() { - ApplicationContext context = new SpringApplicationBuilder( - PojoToPojoStreamListener.class).web(WebApplicationType.NONE).run( - "--spring.cloud.stream.default.contentType=application/x-java-object", - "--spring.jmx.enabled=false"); - InputDestination source = context.getBean(InputDestination.class); - OutputDestination target = context.getBean(OutputDestination.class); - - KryoMessageConverter converter = new KryoMessageConverter(null, true); - @SuppressWarnings("unchecked") - Message message = (Message) converter.toMessage( - new Person("oleg"), - new MessageHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, - MessageConverterUtils.X_JAVA_OBJECT))); - - source.send(new GenericMessage<>(message.getPayload())); - Message outputMessage = target.receive(); - assertThat(outputMessage).isNotNull(); - MimeType contentType = (MimeType) outputMessage.getHeaders() - .get(MessageHeaders.CONTENT_TYPE); - assertThat(contentType.getSubtype()).isEqualTo("x-java-object"); - assertThat(contentType.getParameters().get("type")) - .isEqualTo(Person.class.getName()); - } - - @Test - @SuppressWarnings("deprecation") - public void kryo_pojoToPojoContentTypeHeader() { - ApplicationContext context = new SpringApplicationBuilder( - PojoToPojoStreamListener.class).web(WebApplicationType.NONE).run( - "--spring.jmx.enabled=false", - "--spring.cloud.stream.bindings.output.contentType=application/x-java-object"); - InputDestination source = context.getBean(InputDestination.class); - OutputDestination target = context.getBean(OutputDestination.class); - - KryoMessageConverter converter = new KryoMessageConverter(null, true); - @SuppressWarnings("unchecked") - Message message = (Message) converter.toMessage( - new Person("oleg"), - new MessageHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, - MessageConverterUtils.X_JAVA_OBJECT))); - - source.send(message); - Message outputMessage = target.receive(); - assertThat(outputMessage).isNotNull(); - MimeType contentType = (MimeType) outputMessage.getHeaders() - .get(MessageHeaders.CONTENT_TYPE); - assertThat(contentType.getSubtype()).isEqualTo("x-java-object"); - } - - /** - * This test simply demonstrates how one can override an existing MessageConverter for - * a given contentType. In this case we are demonstrating how Kryo converter can be - * overriden ('application/x-java-object' maps to Kryo). - */ - @Test - public void overrideMessageConverter_defaultContentTypeBinding() { - ApplicationContext context = new SpringApplicationBuilder( - StringToStringStreamListener.class, CustomConverters.class) - .web(WebApplicationType.NONE) - .run("--spring.cloud.stream.default.contentType=application/x-java-object", - "--spring.jmx.enabled=false"); - InputDestination source = context.getBean(InputDestination.class); - OutputDestination target = context.getBean(OutputDestination.class); - String jsonPayload = "{\"name\":\"oleg\"}"; - source.send(new GenericMessage<>(jsonPayload.getBytes())); - Message outputMessage = target.receive(); - assertThat(outputMessage).isNotNull(); - System.out - .println(new String(outputMessage.getPayload(), StandardCharsets.UTF_8)); - assertThat(new String(outputMessage.getPayload(), StandardCharsets.UTF_8)) - .isEqualTo("AlwaysStringKryoMessageConverter"); - assertThat(outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)) - .isEqualTo(MimeType.valueOf("application/x-java-object")); - } - @Test public void customMessageConverter_defaultContentTypeBinding() { ApplicationContext context = new SpringApplicationBuilder( diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/ArgumentResolversTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/ArgumentResolversTests.java index 623b58bca..fec048fba 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/ArgumentResolversTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/ArgumentResolversTests.java @@ -38,7 +38,7 @@ import static org.assertj.core.api.Assertions.assertThat; */ public class ArgumentResolversTests { - @SuppressWarnings({ "deprecation", "rawtypes", "unchecked" }) + @SuppressWarnings({ "rawtypes", "unchecked" }) @Test public void testSmartPayloadArgumentResolver() throws Exception { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BinderConfigurationParsingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BinderConfigurationParsingTests.java index f0a8013f6..82c708a09 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BinderConfigurationParsingTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BinderConfigurationParsingTests.java @@ -51,7 +51,7 @@ public class BinderConfigurationParsingTests { Resource resource = new InputStreamResource( new ByteArrayInputStream(oneBinderConfiguration.getBytes())); - Collection binderConfigurations = BinderFactoryConfiguration + Collection binderConfigurations = BinderFactoryAutoConfiguration .parseBinderConfigurations(classLoader, resource); assertThat(binderConfigurations).isNotNull(); @@ -62,7 +62,6 @@ public class BinderConfigurationParsingTests { .contains(StubBinder1Configuration.class); } - @SuppressWarnings("unchecked") @Test public void testParseTwoBindersConfigurations() throws Exception { // this is just checking that resources are passed and classes are loaded properly @@ -72,7 +71,7 @@ public class BinderConfigurationParsingTests { Resource twoBinderConfigurationResource = new InputStreamResource( new ByteArrayInputStream(binderConfiguration.getBytes())); - Collection twoBinderConfig = BinderFactoryConfiguration + Collection twoBinderConfig = BinderFactoryAutoConfiguration .parseBinderConfigurations(classLoader, twoBinderConfigurationResource); assertThat(twoBinderConfig.size()).isEqualTo(2); @@ -86,7 +85,7 @@ public class BinderConfigurationParsingTests { } private List stubBinders(Collection twoBinderConfigurations, - String binderName, Class... configurationNames) { + String binderName, Class... configurationNames) { return twoBinderConfigurations.stream() .filter(binderType -> binderName.equals(binderType.getDefaultName()) && !Collections.disjoint( @@ -96,7 +95,6 @@ public class BinderConfigurationParsingTests { } @Test - @SuppressWarnings("unchecked") public void testParseTwoBindersWithMultipleClasses() throws Exception { // this is just checking that resources are passed and classes are loaded properly // class values used here are not binder configurations @@ -106,7 +104,7 @@ public class BinderConfigurationParsingTests { Resource binderConfigurationResource = new InputStreamResource( new ByteArrayInputStream(binderConfiguration.getBytes())); - Collection binderConfigurations = BinderFactoryConfiguration + Collection binderConfigurations = BinderFactoryAutoConfiguration .parseBinderConfigurations(classLoader, binderConfigurationResource); assertThat(binderConfigurations.size()).isEqualTo(2); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/converter/KryoMessageConverterTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/converter/KryoMessageConverterTests.java deleted file mode 100644 index c12ffce98..000000000 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/converter/KryoMessageConverterTests.java +++ /dev/null @@ -1,101 +0,0 @@ -/* - * Copyright 2017-2019 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://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.io.ByteArrayOutputStream; - -import com.esotericsoftware.kryo.Kryo; -import com.esotericsoftware.kryo.io.Output; -import org.junit.Test; - -import org.springframework.integration.support.MessageBuilder; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHeaders; -import org.springframework.messaging.converter.MessageConversionException; - -import static org.assertj.core.api.Assertions.assertThat; - -/** - * @author Vinicius Carvalho - */ -public class KryoMessageConverterTests { - - @Test - public void convertStringType() throws Exception { - KryoMessageConverter kryoMessageConverter = new KryoMessageConverter(null, true); - Message message = MessageBuilder.withPayload("foo") - .setHeader(MessageHeaders.CONTENT_TYPE, "application/x-java-object") - .build(); - Message converted = kryoMessageConverter.toMessage(message.getPayload(), - message.getHeaders()); - assertThat(converted).isNotNull(); - assertThat(converted.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString()) - .isEqualTo("application/x-java-object;type=java.lang.String"); - } - - @Test - public void readStringType() throws Exception { - KryoMessageConverter kryoMessageConverter = new KryoMessageConverter(null, true); - Kryo kryo = new Kryo(); - String foo = "foo"; - ByteArrayOutputStream baos = new ByteArrayOutputStream(); - Output output = new Output(baos); - kryo.writeObject(output, foo); - output.close(); - Message message = MessageBuilder.withPayload(baos.toByteArray()) - .setHeader(MessageHeaders.CONTENT_TYPE, - KryoMessageConverter.KRYO_MIME_TYPE + ";type=java.lang.String") - .build(); - Object result = kryoMessageConverter.fromMessage(message, String.class); - assertThat(result).isEqualTo(foo); - } - - @Test - public void testMissingHeaders() throws Exception { - KryoMessageConverter kryoMessageConverter = new KryoMessageConverter(null, true); - Kryo kryo = new Kryo(); - String foo = "foo"; - ByteArrayOutputStream baos = new ByteArrayOutputStream(); - Output output = new Output(baos); - kryo.writeObject(output, foo); - output.close(); - Message message = MessageBuilder.withPayload(baos.toByteArray()).build(); - Object result = kryoMessageConverter.fromMessage(message, String.class); - assertThat(result).isNull(); - } - - @Test(expected = MessageConversionException.class) - public void readWithWrongPayloadType() throws Exception { - KryoMessageConverter kryoMessageConverter = new KryoMessageConverter(null, true); - Message message = MessageBuilder.withPayload("foo") - .setHeader(MessageHeaders.CONTENT_TYPE, - KryoMessageConverter.KRYO_MIME_TYPE + ";type=java.lang.String") - .build(); - kryoMessageConverter.fromMessage(message, String.class); - } - - @Test(expected = MessageConversionException.class) - public void readWithWrongPayloadFormat() throws Exception { - KryoMessageConverter kryoMessageConverter = new KryoMessageConverter(null, true); - Message message = MessageBuilder.withPayload("foo") - .setHeader(MessageHeaders.CONTENT_TYPE, - KryoMessageConverter.KRYO_MIME_TYPE + ";type=java.lang.String") - .build(); - kryoMessageConverter.fromMessage(message, String.class); - } - -} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java index ad0032093..f99e75694 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2017 the original author or authors. + * Copyright 2015-2019 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -28,7 +28,7 @@ import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.BinderFactory; import org.springframework.cloud.stream.binder.ConsumerProperties; -import org.springframework.cloud.stream.config.BinderFactoryConfiguration; +import org.springframework.cloud.stream.config.BinderFactoryAutoConfiguration; import org.springframework.cloud.stream.messaging.Sink; import org.springframework.context.annotation.Import; import org.springframework.context.annotation.PropertySource; @@ -73,7 +73,7 @@ public class PartitionedConsumerTest { @EnableBinding(Sink.class) @EnableAutoConfiguration - @Import({ BinderFactoryConfiguration.class }) + @Import({ BinderFactoryAutoConfiguration.class }) @PropertySource("classpath:/org/springframework/cloud/stream/binder/partitioned-consumer-test.properties") public static class TestSink {