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

GH-1707 Initial removal of 2.0.0 deprecated classes and methods

Browse Source
This commit is contained in:
Oleg Zhurakousky
2019-05-08 08:49:10 +02:00
parent 0075d094fc
commit ee417f9d85
45 changed files with 92 additions and 1559 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
pom.xml
View File

@@ -26,7 +26,6 @@
<spring.tuple.version>1.0.0.RELEASE</spring.tuple.version>
<spring.integration.tuple.version>1.0.0.RELEASE</spring.integration.tuple.version>
<reactor.version>Californium-SR5</reactor.version>
<kryo-shaded.version>3.0.3</kryo-shaded.version>
<objenesis.version>2.1</objenesis.version>
<spring-cloud-function.version>2.1.0.RELEASE</spring-cloud-function.version>
@@ -81,11 +80,6 @@
<artifactId>spring-cloud-stream-test-support-internal</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.esotericsoftware</groupId>
<artifactId>kryo-shaded</artifactId>
<version>${kryo-shaded.version}</version>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-bom</artifactId>

4
spring-cloud-stream-binder-test/pom.xml
View File

@@ -27,6 +27,10 @@
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-web</artifactId>

125
spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractBinderTests.java
View File

@@ -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<B extends AbstractTestBinder<? extends
consumerBinding.unbind();
}
@Test
@SuppressWarnings({ "rawtypes", "deprecation" })
public void testSendAndReceiveKryo() throws Exception {
Binder binder = getBinder();
BindingProperties outputBindingProperties = createProducerBindingProperties(
createProducerProperties());
DirectChannel moduleOutputChannel = createBindableChannel("output",
outputBindingProperties);
BindingProperties inputBindingProperties = createConsumerBindingProperties(
createConsumerProperties());
DirectChannel moduleInputChannel = createBindableChannel("input",
inputBindingProperties);
Binding<MessageChannel> producerBinding = binder.bindProducer(
String.format("foo%s0x", getDestinationNameDelimiter()),
moduleOutputChannel, outputBindingProperties.getProducer());
Binding<MessageChannel> 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<Message<Foo>> inboundMessageRef = new AtomicReference<Message<Foo>>();
moduleInputChannel.subscribe(message1 -> {
try {
inboundMessageRef.set((Message<Foo>) 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<MessageChannel> producerBinding = binder.bindProducer(
String.format("foo%s0y", getDestinationNameDelimiter()),
moduleOutputChannel, outputBindingProperties.getProducer());
Binding<MessageChannel> 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<Message<byte[]>> inboundMessageRef = new AtomicReference<Message<byte[]>>();
moduleInputChannel.subscribe(message1 -> {
try {
inboundMessageRef.set((Message<byte[]>) 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 AbstractBinderTests<B extends AbstractTestBinder<? extends
}
private class Foo {
private String name;
@SuppressWarnings("unused")
public String getName() {
return this.name;
}
public void setName(String name) {
this.name = name;
}
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2017 the original author or authors.
* Copyright 2014-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.
@@ -306,7 +306,6 @@ public abstract class PartitionCapableBinderTests<B extends AbstractTestBinder<?
}
@Test
@SuppressWarnings("deprecation")
public void testPartitionedModuleJava() throws Exception {
B binder = getBinder();
@@ -334,8 +333,6 @@ public abstract class PartitionCapableBinderTests<B extends AbstractTestBinder<?
"testPartitionedModuleJava", input2, consumerProperties);
PP producerProperties = createProducerProperties();
producerProperties.setPartitionKeyExtractorClass(PartitionTestSupport.class);
producerProperties.setPartitionSelectorClass(PartitionTestSupport.class);
producerProperties.setPartitionCount(3);
DirectChannel output = createBindableChannel("output",
createProducerBindingProperties(producerProperties));

5
spring-cloud-stream-integration-tests/pom.xml
View File

@@ -41,6 +41,11 @@
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

84
spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/InboundJsonToTupleConversionTest.java
View File

@@ -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<byte[]> received = (Message<byte[]>) ((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;
}
}
}

27
spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java
View File

@@ -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()

145
spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/contentType/ContentTypeTests.java
View File

@@ -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<User> message = (Message<User>) 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<User> message = (Message<User>) 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<byte[]> message = (Message<byte[]>) 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<String, Object> headers = (Map<String, Object>) 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<String, Object> headers = (Map<String, Object>) 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<String, Object> headers = (Map<String, Object>) 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) {
}

4
spring-cloud-stream-test-support/pom.xml
View File

@@ -29,5 +29,9 @@
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-autoconfigure</artifactId>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
</dependencies>
</project>

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

@@ -37,14 +37,6 @@
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-jmx</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-tuple</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-tuple</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.retry</groupId>
<artifactId>spring-retry</artifactId>

67
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/SharedChannelRegistry.java
View File

@@ -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<String, MessageChannel> getAll() {
Map<String, Object> sharedBindingTargets = this.sharedBindingTargetRegistry
.getAll();
Map<String, MessageChannel> sharedMessageChannels = new HashMap<>();
for (Map.Entry<String, Object> sharedBindingTargetEntry : sharedBindingTargets
.entrySet()) {
if (MessageChannel.class
.isAssignableFrom(sharedBindingTargetEntry.getValue().getClass())) {
sharedMessageChannels.put(sharedBindingTargetEntry.getKey(),
(MessageChannel) sharedBindingTargetEntry.getValue());
}
}
return Collections.unmodifiableMap(sharedMessageChannels);
}
}

48
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Bindings.java
View File

@@ -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();
}

6
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java
View File

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

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

@@ -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<T, C extends ConsumerProperties, P extends
+ (StringUtils.hasText(group) ? group : "default");
}
/**
* Deprecated as of v2.0. Doesn't do anything other then returns an instance of
* {@link MessageValues} built from {@link Message}. Remains primarily for backward
* compatibility and will be removed in the next major release.
* @param message message to serialize
* @return wrapped message
*/
@Deprecated
protected final MessageValues serializePayloadIfNecessary(Message<?> 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.

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

@@ -137,14 +137,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
public AbstractMessageChannelBinder(String[] headersToEmbed,
PP provisioningProvider) {
this(headersToEmbed, provisioningProvider, null);
}
@Deprecated
public AbstractMessageChannelBinder(String[] headersToEmbed, PP provisioningProvider,
ListenerContainerCustomizer<?> 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<C extends ConsumerProperties,
this.delegate.handleMessage(messageToSend);
}
@SuppressWarnings("deprecation")
private Message<?> 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);

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

@@ -22,11 +22,6 @@ package org.springframework.cloud.stream.binder;
*/
public enum HeaderMode {
/**
* @deprecated - use {@link #none}.
*/
raw,
/**
* No headers.
*/

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

@@ -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() {

19
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/Bindable.java
View File

@@ -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.<Binding<Object>>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}

20
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java
View File

@@ -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<Binding<Object>> 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<Binding<Object>> createAndBindOutputs(
BindingService bindingService) {

26
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java
View File

@@ -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<? extends MessageChannel> bindingTargetFactory,
DynamicDestinationsBindable dynamicDestinationsBindable) {
this(bindingService, bindingTargetFactory, dynamicDestinationsBindable, null,
null);
this(bindingService, bindingTargetFactory, dynamicDestinationsBindable, null);
}
@SuppressWarnings("rawtypes")
@@ -64,28 +62,6 @@ public class BinderAwareChannelResolver
AbstractBindingTargetFactory<? extends MessageChannel> 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<? extends MessageChannel> bindingTargetFactory,
DynamicDestinationsBindable dynamicDestinationsBindable,
NewDestinationBindingCallback callback,
GlobalChannelInterceptorProcessor globalChannelInterceptorProcessor) {
this.dynamicDestinationsBindable = dynamicDestinationsBindable;
Assert.notNull(bindingService, "'bindingService' cannot be null");
Assert.notNull(bindingTargetFactory, "'bindingTargetFactory' cannot be null");

9
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareRouterBeanPostProcessor.java → spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareRouter.java
View File

@@ -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<MessageChannel> channelResolver) {
if (routers != null) {
for (AbstractMappingMessageRouter router : routers) {

10
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingBeanDefinitionRegistryUtils.java
View File

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

9
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java
View File

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

45
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java
View File

@@ -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> T instantiate(Class<?> implClass, Class<T> 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.

72
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SingleBindingTargetBindable.java
View File

@@ -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 <T> 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<T> 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<Binding<Object>> 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<String> getOutputs() {
return Collections.unmodifiableSet(new HashSet<>(Arrays.asList(this.name)));
}
}

5
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryConfiguration.java → spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java
View File

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

12
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderProperties.java
View File

@@ -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<String, Object> environment) {
this.environment = environment;

7
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java
View File

@@ -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<MessageChannel> channelResolver) {
return new org.springframework.cloud.stream.binding.BinderAwareRouterBeanPostProcessor(
routers, channelResolver);
return new BinderAwareRouter(routers, channelResolver);
}
@Bean

2
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/SmartMessageMethodArgumentResolver.java
View File

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

3
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/SmartPayloadArgumentResolver.java
View File

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

6
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/SpelExpressionConverterConfiguration.java
View File

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

7
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java
View File

@@ -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));
}
/**

82
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JavaSerializationMessageConverter.java
View File

@@ -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();
}
}

74
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonUnmarshallingConverter.java
View File

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

265
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/KryoMessageConverter.java
View File

@@ -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<MimeType> supportedMimeTypes;
private ConcurrentMap<String, MimeType> mimeTypesCache = new ConcurrentHashMap<>();
private ContentTypeResolver contentTypeResolver = new DefaultContentTypeResolver();
public KryoMessageConverter(List<KryoRegistrar> 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> T deserialize(byte[] bytes, Class<T> 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> T deserialize(InputStream inputStream, final Class<T> 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<MimeType> getSupportedMimeTypes() {
return this.supportedMimeTypes;
}
}

1
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/MessageConverterUtils.java
View File

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

101
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/TupleJsonMessageConverter.java
View File

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

13
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
View File

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

11
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryConfigurationTests.java → spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryAutoConfigurationTests.java
View File

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

2
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/HealthIndicatorsConfigurationTests.java
View File

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

80
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java
View File

@@ -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<byte[]> message = (Message<byte[]>) converter.toMessage(
new Person("oleg"),
new MessageHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE,
MessageConverterUtils.X_JAVA_OBJECT)));
source.send(new GenericMessage<>(message.getPayload()));
Message<byte[]> 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<byte[]> message = (Message<byte[]>) converter.toMessage(
new Person("oleg"),
new MessageHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE,
MessageConverterUtils.X_JAVA_OBJECT)));
source.send(message);
Message<byte[]> 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<byte[]> 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(

2
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/ArgumentResolversTests.java
View File

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

10
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BinderConfigurationParsingTests.java
View File

@@ -51,7 +51,7 @@ public class BinderConfigurationParsingTests {
Resource resource = new InputStreamResource(
new ByteArrayInputStream(oneBinderConfiguration.getBytes()));
Collection<BinderType> binderConfigurations = BinderFactoryConfiguration
Collection<BinderType> 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<BinderType> twoBinderConfig = BinderFactoryConfiguration
Collection<BinderType> twoBinderConfig = BinderFactoryAutoConfiguration
.parseBinderConfigurations(classLoader, twoBinderConfigurationResource);
assertThat(twoBinderConfig.size()).isEqualTo(2);
@@ -86,7 +85,7 @@ public class BinderConfigurationParsingTests {
}
private List<BinderType> stubBinders(Collection<BinderType> 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<BinderType> binderConfigurations = BinderFactoryConfiguration
Collection<BinderType> binderConfigurations = BinderFactoryAutoConfiguration
.parseBinderConfigurations(classLoader, binderConfigurationResource);
assertThat(binderConfigurations.size()).isEqualTo(2);

101
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/converter/KryoMessageConverterTests.java
View File

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

6
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java
View File

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