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 ef339e771..50b74b312 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 @@ -85,9 +85,9 @@ public abstract class AbstractBinderTests producerBinding = binder.bindProducer(String.format("foo%s0y", getDestinationNameDelimiter()), moduleOutputChannel, outputBindingProperties.getProducer()); - + Binding consumerBinding = binder.bindConsumer(String.format("foo%s0y", getDestinationNameDelimiter()), "testSendAndReceiveJavaSerialization", moduleInputChannel, inputBindingProperties.getConsumer()); diff --git a/spring-cloud-stream-core-docs/src/main/asciidoc/preface.adoc b/spring-cloud-stream-core-docs/src/main/asciidoc/preface.adoc index 3ffa4bcb8..928a8951e 100644 --- a/spring-cloud-stream-core-docs/src/main/asciidoc/preface.adoc +++ b/spring-cloud-stream-core-docs/src/main/asciidoc/preface.adoc @@ -18,12 +18,12 @@ what messaging midleware this application will be bound to. Choose the one you h Also, as you can see from the Initilaizer screen there are few other options you can choose. For example, you can choose Gradle as your build tool instead of the default Maven. With the _Dependencies_ selected the only other thing you have to identify is the application name - _logging-consumer_. Your configuration screeen should now contain the following: - + Dependencies: Cloud Stream, RabbitMQ (or Kafka) Group: com.example - default Artifact: logging-consumer Spring Boot Version: 2.0.0 (or above) - default - + Click on _Generate Project_ button. This will donwload the zipped version of the generated project to your hard drive. Unzip it and you're ready for Step Two. ==== Step Two - Import project into the IDE @@ -47,12 +47,12 @@ public class LoggingConsumerApplication { public static void main(String[] args) { SpringApplication.run(LoggingConsumerApplication.class, args); } - + @StreamListener(Sink.INPUT) public void handle(Person person) { System.out.println("Received: " + person); } - + public static class Person { private String name; public String getName() { @@ -99,7 +99,7 @@ The contents of the message should be JSON representation of `Person` class, so And in your console you should see: Received: Turd Ferguson - + You can also build/package your application into a boot jar (i.e., `./mvnw clean install`) and run the built JAR using `java -jar` command. That is all! 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 ac4cf7ca3..f4f6b2b87 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 @@ -228,7 +228,7 @@ public class ContentTypeTests { assertThat(message.getPayload()).isEqualTo(user.toString()); } } - + @Test public void testSendTuple() throws Exception { try (ConfigurableApplicationContext context = SpringApplication.run( diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterAutoConfiguration.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterAutoConfiguration.java index a6231d496..bdc564e5e 100644 --- a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterAutoConfiguration.java +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterAutoConfiguration.java @@ -65,7 +65,7 @@ public class AvroMessageConverterAutoConfiguration { this.avroMessageConverterProperties.getSchemaLocations()); } avroSchemaRegistryClientMessageConverter.setPrefix(this.avroMessageConverterProperties.getPrefix()); - + try { Class clazz = this.avroMessageConverterProperties.getSubjectNamingStrategy(); Constructor constructor = ReflectionUtils.accessibleConstructor(clazz); diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java index 8085e36b2..23c1b3536 100644 --- a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java @@ -89,7 +89,7 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag public static final String REFERENCE_CACHE_NAME = CACHE_PREFIX + ".referenceCache"; public static final MimeType DEFAULT_AVRO_MIME_TYPE = new MimeType("application", "*+" + AVRO_FORMAT); - + private Pattern versionedSchema; private boolean dynamicSchemaGenerationEnabled; @@ -250,7 +250,7 @@ public class AvroSchemaRegistryClientMessageConverter extends AbstractAvroMessag _headers.put(MessageHeaders.CONTENT_TYPE, "application/" + this.prefix + "." + schemaReference.getSubject() + ".v" + schemaReference.getVersion() + "+" + AVRO_FORMAT); - + return schema; } diff --git a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/DefaultSchemaRegistryClient.java b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/DefaultSchemaRegistryClient.java index 5ab12a093..c43b18e8d 100644 --- a/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/DefaultSchemaRegistryClient.java +++ b/spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/DefaultSchemaRegistryClient.java @@ -38,12 +38,12 @@ public class DefaultSchemaRegistryClient implements SchemaRegistryClient { public DefaultSchemaRegistryClient() { this(new RestTemplate()); } - + public DefaultSchemaRegistryClient(RestTemplate restTemplate) { Assert.notNull(restTemplate,"'restTemplate' must not be null."); this.restTemplate = restTemplate; } - + protected String getEndpoint() { return this.endpoint; } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java index fc526ac2a..73894bf88 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java @@ -36,11 +36,11 @@ import org.springframework.integration.endpoint.Pausable; * @see org.springframework.cloud.stream.annotation.EnableBinding */ public interface Binding extends Pausable { - + default Map getExtendedInfo() { return Collections.emptyMap(); } - + /** * Stops the target component represented by this instance. * NOTE: At the time the instance is created the component is already started. @@ -49,7 +49,7 @@ public interface Binding extends Pausable { * @see BindingsEndpoint */ default void start() {} - + /** * Starts the target component represented by this instance. * NOTE: At the time the instance is created the component is already started. @@ -58,7 +58,7 @@ public interface Binding extends Pausable { * @see BindingsEndpoint */ default void stop() {} - + /** * Pauses the target component represented by this instance if and only if the component * implements {@link Pausable} interface @@ -70,7 +70,7 @@ public interface Binding extends Pausable { default void pause() { this.stop(); } - + /** * Resumes the target component represented by this instance if and only if the component * implements {@link Pausable} interface @@ -82,14 +82,14 @@ public interface Binding extends Pausable { default void resume() { this.start(); } - + /** * Returns 'true' if the target component represented by this instance is running. */ default boolean isRunning() { return false; } - + /** * Returns the name of this binding (i.e., channel name) * @@ -98,7 +98,7 @@ public interface Binding extends Pausable { default String getName() { return null; } - + /** * Unbinds the target component represented by this instance and stops any active diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java index 950706eac..cbe737ac9 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java @@ -165,7 +165,7 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl Assert.state(binderConfiguration != null, "Unknown binder configuration: " + configurationName); BinderType binderType = this.binderTypeRegistry.get(binderConfiguration.getBinderType()); Assert.notNull(binderType, "Binder type " + binderConfiguration.getBinderType() + " is not defined"); - + Map binderProperties = new HashMap<>(); this.flatten(null, binderConfiguration.getProperties(), binderProperties); @@ -214,7 +214,7 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl } return (Binder) this.binderInstanceCache.get(configurationName).getKey(); } - + /** * Ensures that nested properties are flattened (i.e., foo.bar=baz instead of foo={bar=baz}) */ diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java index b624a88ea..c3da089ef 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java @@ -42,7 +42,7 @@ import org.springframework.util.StringUtils; @JsonPropertyOrder({ "name", "group", "pausable", "state"}) @JsonIgnoreProperties("running") public class DefaultBinding implements Binding { - + private final Log logger = LogFactory.getLog(this.getClass().getName()); protected final String name; @@ -52,9 +52,9 @@ public class DefaultBinding implements Binding { protected final T target; protected final Lifecycle lifecycle; - + private boolean paused; - + /** * Creates an instance that associates a given name, group and binding target with an * optional {@link Lifecycle} component, which will be stopped during unbinding. @@ -73,7 +73,7 @@ public class DefaultBinding implements Binding { this.target = target; this.lifecycle = lifecycle; } - + public String getName() { return this.name; } @@ -81,7 +81,7 @@ public class DefaultBinding implements Binding { public String getGroup() { return this.group; } - + public String getState() { String state = "N/A"; if (this.lifecycle != null) { @@ -94,15 +94,15 @@ public class DefaultBinding implements Binding { } return state; } - + public boolean isRunning() { return this.lifecycle != null && this.lifecycle.isRunning(); } - + public boolean isPausable() { return this.lifecycle instanceof Pausable; } - + @Override public final synchronized void start() { if (!this.isRunning()) { @@ -114,14 +114,14 @@ public class DefaultBinding implements Binding { } } } - + @Override public final synchronized void stop() { if (this.isRunning()) { this.lifecycle.stop(); } } - + @Override public final synchronized void pause() { if (this.lifecycle instanceof Pausable) { @@ -132,7 +132,7 @@ public class DefaultBinding implements Binding { logger.warn("Attempted to pause a component that does not support Pausable " + this.lifecycle); } } - + @Override public final synchronized void resume() { if (this.lifecycle instanceof Pausable) { @@ -162,14 +162,14 @@ public class DefaultBinding implements Binding { : ObjectUtils.nullSafeToString(this.lifecycle)) + "]"; } - + /** * Listener method that executes after unbinding. Subclasses can implement their own * behaviour on unbinding by overriding this method. */ protected void afterUnbind() { } - + private String getRunningState() { return isRunning() ? "running" : "stopped"; } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java index 5452e0356..8eafef39c 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java @@ -177,7 +177,7 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life public boolean poll(MessageHandler handler) { return poll(handler, null); } - + @Override public boolean poll(MessageHandler handler, ParameterizedTypeReference type) { Message message = this.receive(type); @@ -249,7 +249,7 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life */ private Message receive(ParameterizedTypeReference type) { Message message = this.source.receive(); - if (message != null && type != null && this.messageConverter != null) { + if (message != null && type != null && this.messageConverter != null) { Class targetType = type == null ? Object.class : type.getType() instanceof Class ? (Class) type.getType() : Object.class; Object payload = this.messageConverter.fromMessage(message, targetType, type); @@ -262,7 +262,7 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life } return message; } - + private void doHandleMessage(MessageHandler handler, Message message) { try { handler.handleMessage(message); @@ -271,7 +271,7 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life throw new MessageHandlingException(message, t); } } - + /** * If there's a retry template, it will set the attributes holder via the listener. If * there's no retry template, but there's an error channel, we create a new attributes @@ -295,7 +295,7 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life } } } - + private void handle(Message message, MessageHandler handler) { setAttributesIfNecessary(message); doHandleMessage(handler, message); 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 818c1840e..8f4e7dc3c 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 @@ -40,7 +40,7 @@ public interface Bindable { default void bindInputs(BindingService adapter) { this.createAndBindInputs(adapter); } - + /** * Binds all the inputs 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 ebe078b1b..7164afd24 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 @@ -217,7 +217,7 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean> createAndBindInputs(BindingService bindingService) { List> bindings = new ArrayList<>(); 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 0f851e706..5ec8afee5 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 @@ -43,7 +43,7 @@ import org.springframework.util.ObjectUtils; * @author Oleg Zhurakousky */ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestinationResolver { - + private final Log logger = LogFactory.getLog(BinderAwareChannelResolver.class); private final BindingService bindingService; @@ -56,15 +56,15 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina private final NewDestinationBindingCallback newBindingCallback; private ConfigurableListableBeanFactory beanFactory; - + private final GlobalChannelInterceptorProcessor globalChannelInterceptorProcessor; - + public BinderAwareChannelResolver(BindingService bindingService, AbstractBindingTargetFactory bindingTargetFactory, DynamicDestinationsBindable dynamicDestinationsBindable) { this(bindingService, bindingTargetFactory, dynamicDestinationsBindable, null, null); } - + @SuppressWarnings("rawtypes") public BinderAwareChannelResolver(BindingService bindingService, AbstractBindingTargetFactory bindingTargetFactory, @@ -114,12 +114,12 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina throw e; } } - + MessageChannel channel = this.bindingTargetFactory.createOutput(channelName); this.beanFactory.registerSingleton(channelName, channel); - + this.instrumentChannelWithGlobalInterceptors(channel, channelName); - + channel = (MessageChannel) this.beanFactory.initializeBean(channel, channelName); if (this.newBindingCallback != null) { ProducerProperties producerProperties = bindingServiceProperties.getProducerProperties(channelName); @@ -129,16 +129,16 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina } Binding binding = this.bindingService.bindProducer(channel, channelName); this.dynamicDestinationsBindable.addOutputBinding(channelName, binding); - + return channel; } } - + private void instrumentChannelWithGlobalInterceptors(MessageChannel channel, String channelName) { if (channel instanceof ChannelInterceptorAware) { if (this.globalChannelInterceptorProcessor != null) { this.globalChannelInterceptorProcessor.addMatchingInterceptors((ChannelInterceptorAware) channel, channelName); - } + } } else { logger.warn("Failed to add global interceptors to '" + channelName + "' since it is not an instance of ChannelInterceptorAware."); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/InputBindingLifecycle.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/InputBindingLifecycle.java index 57f2f84e8..6b5e0f912 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/InputBindingLifecycle.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/InputBindingLifecycle.java @@ -29,7 +29,7 @@ import org.springframework.cloud.stream.binder.Binding; * @author Oleg Zhurakousky */ public class InputBindingLifecycle extends AbstractBindingLifecycle { - + @SuppressWarnings("unused") //It is actually used reflectively since at the moment we do not want to expose it via public method private Collection> inputBindings; diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageSourceBindingTargetFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageSourceBindingTargetFactory.java index 53d386e11..03c5aec4c 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageSourceBindingTargetFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageSourceBindingTargetFactory.java @@ -31,7 +31,7 @@ public class MessageSourceBindingTargetFactory extends AbstractBindingTargetFactory { private final MessageChannelAndSourceConfigurer messageSourceConfigurer; - + private final SmartMessageConverter messageConverter; public MessageSourceBindingTargetFactory(SmartMessageConverter messageConverter, MessageChannelConfigurer messageSourceConfigurer) { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java index 3292b4ac8..b2c2662f6 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java @@ -38,7 +38,7 @@ import org.springframework.validation.annotation.Validated; @JsonInclude(Include.NON_DEFAULT) @Validated public class BindingProperties { - + public static final MimeType DEFAULT_CONTENT_TYPE = MimeTypeUtils.APPLICATION_JSON; private static final String COMMA = ","; @@ -81,7 +81,7 @@ public class BindingProperties { */ private ProducerProperties producer; - + public String getDestination() { return this.destination; } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java index 320ddbebc..fb5397a90 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java @@ -163,9 +163,9 @@ public class BindingServiceProperties implements ApplicationContextAware, Initia GenericConversionService cs = (GenericConversionService) IntegrationUtils.getConversionService(this.applicationContext.getBeanFactory()); if (this.applicationContext.containsBean("spelConverter")) { Converter converter = (Converter) this.applicationContext.getBean("spelConverter"); - cs.addConverter(converter); + cs.addConverter(converter); } - + if (this.applicationContext.getEnvironment() instanceof ConfigurableEnvironment) { // override the bindings store with the environment-initializing version if in // a Spring context diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingsEndpointAutoConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingsEndpointAutoConfiguration.java index 8e9f6153e..025c7e395 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingsEndpointAutoConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingsEndpointAutoConfiguration.java @@ -39,7 +39,7 @@ import org.springframework.context.annotation.Configuration; @AutoConfigureAfter(EndpointAutoConfiguration.class) public class BindingsEndpointAutoConfiguration { - @Bean + @Bean public BindingsEndpoint bindingsEndpoint(List inputBindings) { return new BindingsEndpoint(inputBindings); } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java index fd95b919d..dd9cdc765 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java @@ -45,15 +45,15 @@ import org.springframework.messaging.converter.MessageConversionException; * */ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageConverter { - + private final Map, JavaType> typeCache = new ConcurrentHashMap<>(); - + ApplicationJsonMessageMarshallingConverter(@Nullable ObjectMapper objectMapper) { if (objectMapper != null) { this.setObjectMapper(objectMapper); - } + } } - + @Override protected Object convertToInternal(Object payload, @Nullable MessageHeaders headers, @Nullable Object conversionHint) { if (payload instanceof byte[]) { @@ -88,7 +88,7 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC else if (conversionHint instanceof ParameterizedTypeReference) { result = convertParameterizedType(message, targetClass, (ParameterizedTypeReference)conversionHint); } - + if (result == null) { if (message.getPayload() instanceof byte[] && targetClass.isAssignableFrom(String.class)) { result = new String((byte[])message.getPayload(), StandardCharsets.UTF_8); @@ -97,10 +97,10 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC result = super.convertFromInternal(message, targetClass, conversionHint); } } - + return result; } - + private Object convertParameterizedType(Message message, Class targetClass, ParameterizedTypeReference conversionHint) { ObjectMapper objectMapper = this.getObjectMapper(); Object payload = message.getPayload(); 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 6af4acd2c..53fd3c29e 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 @@ -68,7 +68,7 @@ public class CompositeMessageConverterFactory { this.converters = new ArrayList<>(); } initDefaultConverters(); - + DefaultContentTypeResolver resolver = new DefaultContentTypeResolver(); resolver.setDefaultMimeType(BindingProperties.DEFAULT_CONTENT_TYPE); this.converters.stream() @@ -80,11 +80,11 @@ public class CompositeMessageConverterFactory { private void initDefaultConverters() { ApplicationJsonMessageMarshallingConverter applicationJsonConverter = new ApplicationJsonMessageMarshallingConverter(this.objectMapper); applicationJsonConverter.setStrictContentTypeMatch(true); - this.converters.add(applicationJsonConverter); + this.converters.add(applicationJsonConverter); this.converters.add(new TupleJsonMessageConverter(this.objectMapper)); this.converters.add(new ByteArrayMessageConverter()); this.converters.add(new ObjectStringMessageConverter()); - + // Deprecated converters this.converters.add(new JavaSerializationMessageConverter()); this.converters.add(new KryoMessageConverter(null,true)); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/endpoint/BindingsEndpoint.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/endpoint/BindingsEndpoint.java index 9992ff0af..ba4b4ba39 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/endpoint/BindingsEndpoint.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/endpoint/BindingsEndpoint.java @@ -42,17 +42,17 @@ import org.springframework.util.Assert; */ @Endpoint(id = "bindings") public class BindingsEndpoint { - + private final List inputBindingLifecycles; - + private final ObjectMapper objectMapper; - + public BindingsEndpoint(List inputBindingLifecycles) { Assert.notEmpty(inputBindingLifecycles, "'inputBindingLifecycles' must not be null or empty"); this.inputBindingLifecycles = inputBindingLifecycles; this.objectMapper = new ObjectMapper(); } - + @WriteOperation public void changeState(@Selector String name, State state) { Binding binding = BindingsEndpoint.this.locateBinding(name); @@ -75,18 +75,18 @@ public class BindingsEndpoint { } } } - + @ReadOperation public List queryStates() { - return objectMapper.convertValue(gatherInputBindings(), List.class); + return objectMapper.convertValue(gatherInputBindings(), List.class); } - + @ReadOperation public Binding queryState(@Selector String name) { Assert.notNull(name, "'name' must not be null"); - return this.locateBinding(name); + return this.locateBinding(name); } - + @SuppressWarnings("unchecked") private List> gatherInputBindings() { List> inputBindings = new ArrayList<>(); @@ -97,14 +97,14 @@ public class BindingsEndpoint { } return inputBindings; } - + private Binding locateBinding(String name) { return BindingsEndpoint.this.gatherInputBindings().stream() .filter(binding -> name.equals(binding.getName())) .findFirst() .orElse(null); } - + private enum State { STARTED, STOPPED, diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/ApplicationMetricsProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/ApplicationMetricsProperties.java index 98cf47bf8..e8e6fc3fb 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/ApplicationMetricsProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/ApplicationMetricsProperties.java @@ -43,7 +43,7 @@ import org.springframework.util.PatternMatchUtils; */ @ConfigurationProperties(prefix = ApplicationMetricsProperties.PREFIX) public class ApplicationMetricsProperties implements EnvironmentAware, ApplicationContextAware { - + public static final String PREFIX = "spring.cloud.stream.metrics"; private static final Bindable> STRING_STRING_MAP = Bindable.mapOf(String.class, String.class); @@ -60,13 +60,13 @@ public class ApplicationMetricsProperties implements EnvironmentAware, Applicati * For example: `spring.application**` */ private String[] properties; - + /** * Interval expressed as Duration for scheduling metrics snapshots publishing. * Defaults to PT60S (60 sec) */ private String scheduleInterval; - + /** * List of properties that are going to be appended to each message. This gets * populate by onApplicationEvent, once the context refreshes to avoid overhead of @@ -110,7 +110,7 @@ public class ApplicationMetricsProperties implements EnvironmentAware, Applicati } return this.exportProperties; } - + public String getScheduleInterval() { return scheduleInterval; } @@ -165,5 +165,5 @@ public class ApplicationMetricsProperties implements EnvironmentAware, Applicati } return target; } - + } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/MetricsPublisherConfig.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/MetricsPublisherConfig.java index 0fd19be25..746f12bd0 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/MetricsPublisherConfig.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/MetricsPublisherConfig.java @@ -26,7 +26,7 @@ import io.micrometer.core.instrument.step.StepRegistryConfig; * */ class MetricsPublisherConfig implements StepRegistryConfig { - + private final ApplicationMetricsProperties applicationMetricsProperties; MetricsPublisherConfig(ApplicationMetricsProperties applicationMetricsProperties) { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java index 5efc7735b..05d019b13 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java @@ -90,13 +90,13 @@ public class BinderAwareChannelResolverTests { return new GlobalChannelInterceptorWrapper(new ImmutableMessageChannelInterceptor()); } } - + @SuppressWarnings("unchecked") @Before public void setupContext() throws Exception { this.context = new SpringApplicationBuilder(TestChannelBinderConfiguration.getCompleteConfiguration(BinderAwareChannelResolverTests.InterceptorConfiguration.class)) .web(WebApplicationType.NONE).run(); - + this.resolver = context.getBean(BinderAwareChannelResolver.class); this.binder = context.getBean(Binder.class); this.bindingServiceProperties = context.getBean(BindingServiceProperties.class); 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 24c487f28..9fb34cb80 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 @@ -70,7 +70,7 @@ import static org.junit.Assert.assertTrue; * */ public class ContentTypeTckTests { - + @Test public void withInternalPipeline() { ApplicationContext context = new SpringApplicationBuilder(InternalPipeLine.class) @@ -83,7 +83,7 @@ public class ContentTypeTckTests { Message outputMessage = target.receive(); assertEquals("OLEG", new String(outputMessage.getPayload())); } - + @Test public void pojoToPojo() { ApplicationContext context = new SpringApplicationBuilder(PojoToPojoStreamListener.class) @@ -97,7 +97,7 @@ public class ContentTypeTckTests { assertEquals(MimeTypeUtils.APPLICATION_JSON, outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)); assertEquals(jsonPayload, new String(outputMessage.getPayload(), StandardCharsets.UTF_8)); } - + @Test public void pojoToString() { ApplicationContext context = new SpringApplicationBuilder(PojoToStringStreamListener.class) @@ -111,7 +111,7 @@ public class ContentTypeTckTests { assertEquals(MimeTypeUtils.APPLICATION_JSON, outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)); assertEquals("oleg", new String(outputMessage.getPayload(), StandardCharsets.UTF_8)); } - + @Test public void pojoToStringOutboundContentTypeBinding() { ApplicationContext context = new SpringApplicationBuilder(PojoToStringStreamListener.class) @@ -125,7 +125,7 @@ public class ContentTypeTckTests { assertEquals(MimeTypeUtils.TEXT_PLAIN, outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)); assertEquals("oleg", new String(outputMessage.getPayload(), StandardCharsets.UTF_8)); } - + @Test public void pojoToByteArray() { ApplicationContext context = new SpringApplicationBuilder(PojoToByteArrayStreamListener.class) @@ -139,7 +139,7 @@ public class ContentTypeTckTests { assertEquals(MimeTypeUtils.APPLICATION_JSON, outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)); assertEquals("oleg", new String(outputMessage.getPayload(), StandardCharsets.UTF_8)); } - + @Test public void pojoToByteArrayOutboundContentTypeBinding() { ApplicationContext context = new SpringApplicationBuilder(PojoToByteArrayStreamListener.class) @@ -153,7 +153,7 @@ public class ContentTypeTckTests { assertEquals(MimeTypeUtils.TEXT_PLAIN, outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)); assertEquals("oleg", new String(outputMessage.getPayload(), StandardCharsets.UTF_8)); } - + @Test public void stringToPojoInboundContentTypeBinding() { ApplicationContext context = new SpringApplicationBuilder(StringToPojoStreamListener.class) @@ -167,7 +167,7 @@ public class ContentTypeTckTests { assertEquals(MimeTypeUtils.APPLICATION_JSON, outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)); assertEquals(jsonPayload, new String(outputMessage.getPayload(), StandardCharsets.UTF_8)); } - + @Test public void stringToPojoInboundContentTypeHeader() { ApplicationContext context = new SpringApplicationBuilder(StringToPojoStreamListener.class) @@ -181,7 +181,7 @@ public class ContentTypeTckTests { assertEquals(MimeTypeUtils.APPLICATION_JSON, outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)); assertEquals(jsonPayload, new String(outputMessage.getPayload(), StandardCharsets.UTF_8)); } - + @Test public void byteArrayToPojoInboundContentTypeBinding() { ApplicationContext context = new SpringApplicationBuilder(ByteArrayToPojoStreamListener.class) @@ -195,7 +195,7 @@ public class ContentTypeTckTests { assertEquals(MimeTypeUtils.APPLICATION_JSON, outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)); assertEquals(jsonPayload, new String(outputMessage.getPayload(), StandardCharsets.UTF_8)); } - + @Test public void byteArrayToPojoInboundContentTypeHeader() { ApplicationContext context = new SpringApplicationBuilder(StringToPojoStreamListener.class) @@ -209,7 +209,7 @@ public class ContentTypeTckTests { assertEquals(MimeTypeUtils.APPLICATION_JSON, outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)); assertEquals(jsonPayload, new String(outputMessage.getPayload(), StandardCharsets.UTF_8)); } - + @Test public void byteArrayToByteArray() { ApplicationContext context = new SpringApplicationBuilder(ByteArrayToByteArrayStreamListener.class) @@ -223,7 +223,7 @@ public class ContentTypeTckTests { assertEquals(MimeTypeUtils.APPLICATION_JSON, outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)); assertEquals(jsonPayload, new String(outputMessage.getPayload(), StandardCharsets.UTF_8)); } - + @Test public void byteArrayToByteArrayInboundOutboundContentTypeBinding() { ApplicationContext context = new SpringApplicationBuilder(ByteArrayToByteArrayStreamListener.class) @@ -237,8 +237,8 @@ public class ContentTypeTckTests { assertEquals(MimeTypeUtils.TEXT_PLAIN, outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)); assertEquals(jsonPayload, new String(outputMessage.getPayload(), StandardCharsets.UTF_8)); } - - + + @Test public void pojoMessageToStringMessage() { ApplicationContext context = new SpringApplicationBuilder(PojoMessageToStringMessageStreamListener.class) @@ -252,7 +252,7 @@ public class ContentTypeTckTests { assertEquals(MimeTypeUtils.TEXT_PLAIN, outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)); assertEquals("oleg", new String(outputMessage.getPayload(), StandardCharsets.UTF_8)); } - + @Test public void pojoMessageToStringMessageServiceActivator() { ApplicationContext context = new SpringApplicationBuilder(PojoMessageToStringMessageServiceActivator.class) @@ -266,7 +266,7 @@ public class ContentTypeTckTests { assertEquals(MimeTypeUtils.TEXT_PLAIN, outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)); assertEquals("oleg", new String(outputMessage.getPayload(), StandardCharsets.UTF_8)); } - + @Test public void byteArrayMessageToStringJsonMessageStreamListener() { ApplicationContext context = new SpringApplicationBuilder(ByteArrayMessageToStringJsonMessageStreamListener.class) @@ -280,7 +280,7 @@ public class ContentTypeTckTests { assertEquals(MimeTypeUtils.APPLICATION_JSON, outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)); assertEquals("{\"name\":\"bob\"}", new String(outputMessage.getPayload(), StandardCharsets.UTF_8)); } - + @Test public void byteArrayMessageToStringMessageStreamListener() { ApplicationContext context = new SpringApplicationBuilder(StringMessageToStringMessageStreamListener.class) @@ -294,7 +294,7 @@ public class ContentTypeTckTests { assertEquals(MimeTypeUtils.TEXT_PLAIN, outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)); assertEquals("oleg", new String(outputMessage.getPayload(), StandardCharsets.UTF_8)); } - + @Test public void kryo_pojoToPojo() { ApplicationContext context = new SpringApplicationBuilder(PojoToPojoStreamListener.class) @@ -302,12 +302,12 @@ public class ContentTypeTckTests { .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(); assertNotNull(outputMessage); @@ -315,7 +315,7 @@ public class ContentTypeTckTests { assertEquals("x-java-object", contentType.getSubtype()); assertEquals(Person.class.getName(), contentType.getParameters().get("type")); } - + @Test public void kryo_pojoToPojoContentTypeHeader() { ApplicationContext context = new SpringApplicationBuilder(PojoToPojoStreamListener.class) @@ -323,19 +323,19 @@ public class ContentTypeTckTests { .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(); assertNotNull(outputMessage); MimeType contentType = (MimeType) outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE); assertEquals("x-java-object", contentType.getSubtype()); } - + /** * 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). @@ -355,7 +355,7 @@ public class ContentTypeTckTests { assertEquals("AlwaysStringKryoMessageConverter", new String(outputMessage.getPayload(), StandardCharsets.UTF_8)); assertEquals(MimeType.valueOf("application/x-java-object"), outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)); } - + @Test public void customMessageConverter_defaultContentTypeBinding() { ApplicationContext context = new SpringApplicationBuilder(StringToStringStreamListener.class, CustomConverters.class) @@ -371,21 +371,21 @@ public class ContentTypeTckTests { assertEquals(MimeType.valueOf("foo/bar"), outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)); } - + //Failure tests - + @Test public void _jsonToPojoWrongDefaultContentTypeProperty() { ApplicationContext context = new SpringApplicationBuilder(PojoToPojoStreamListener.class) .web(WebApplicationType.NONE) - .run("--spring.cloud.stream.default.contentType=text/plain", "--spring.jmx.enabled=false"); + .run("--spring.cloud.stream.default.contentType=text/plain", "--spring.jmx.enabled=false"); InputDestination source = context.getBean(InputDestination.class); TestChannelBinder binder = context.getBean(TestChannelBinder.class); String jsonPayload = "{\"name\":\"oleg\"}"; source.send(new GenericMessage(jsonPayload.getBytes())); assertTrue(binder.getLastError().getPayload() instanceof MessageConversionException); } - + @Test public void _toStringDefaultContentTypePropertyUnknownContentType() { ApplicationContext context = new SpringApplicationBuilder(StringToStringStreamListener.class) @@ -397,10 +397,10 @@ public class ContentTypeTckTests { source.send(new GenericMessage(jsonPayload.getBytes())); assertTrue(binder.getLastError().getPayload() instanceof MessageConversionException); } - - + + @Test - public void toCollectionWithParameterizedType() throws Exception { + public void toCollectionWithParameterizedType() throws Exception { ApplicationContext context = new SpringApplicationBuilder(CollectionWithParameterizedTypes.class) .web(WebApplicationType.NONE) .run("--spring.jmx.enabled=false"); @@ -411,7 +411,7 @@ public class ContentTypeTckTests { Message outputMessage = target.receive(); assertThat(outputMessage.getPayload()).isEqualTo(jsonPayload.getBytes()); } - + @EnableBinding(Processor.class) @Import(TestChannelBinderConfiguration.class) public static class CollectionWithParameterizedTypes { @@ -422,7 +422,7 @@ public class ContentTypeTckTests { return value; } } - + @EnableBinding(Processor.class) @Import(TestChannelBinderConfiguration.class) public static class TextInJsonOutListener { @@ -432,7 +432,7 @@ public class ContentTypeTckTests { return MessageBuilder.withPayload(value).setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON).build(); } } - + @EnableBinding(Processor.class) @Import(TestChannelBinderConfiguration.class) public static class PojoToPojoStreamListener { @@ -442,7 +442,7 @@ public class ContentTypeTckTests { return value; } } - + @EnableBinding(Processor.class) @Import(TestChannelBinderConfiguration.class) public static class PojoToStringStreamListener { @@ -452,7 +452,7 @@ public class ContentTypeTckTests { return value.toString(); } } - + @EnableBinding(Processor.class) @Import(TestChannelBinderConfiguration.class) public static class PojoToByteArrayStreamListener { @@ -462,7 +462,7 @@ public class ContentTypeTckTests { return value.toString().getBytes(StandardCharsets.UTF_8); } } - + @EnableBinding(Processor.class) @Import(TestChannelBinderConfiguration.class) public static class ByteArrayToPojoStreamListener { @@ -473,7 +473,7 @@ public class ContentTypeTckTests { return mapper.readValue(value, Person.class); } } - + @EnableBinding(Processor.class) @Import(TestChannelBinderConfiguration.class) public static class StringToPojoStreamListener { @@ -484,7 +484,7 @@ public class ContentTypeTckTests { return mapper.readValue(value, Person.class); } } - + @EnableBinding(Processor.class) @Import(TestChannelBinderConfiguration.class) public static class ByteArrayToByteArrayStreamListener { @@ -494,7 +494,7 @@ public class ContentTypeTckTests { return value; } } - + @EnableBinding(Processor.class) @Import(TestChannelBinderConfiguration.class) public static class StringToStringStreamListener { @@ -504,7 +504,7 @@ public class ContentTypeTckTests { return value; } } - + @EnableBinding(Processor.class) @Import(TestChannelBinderConfiguration.class) public static class PojoMessageToStringMessageStreamListener { @@ -514,7 +514,7 @@ public class ContentTypeTckTests { return MessageBuilder.withPayload(value.getPayload().toString()).setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN).build(); } } - + @EnableBinding(Processor.class) @Import(TestChannelBinderConfiguration.class) public static class PojoMessageToStringMessageServiceActivator { @@ -523,7 +523,7 @@ public class ContentTypeTckTests { return MessageBuilder.withPayload(value.getPayload().toString()).setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN).build(); } } - + @EnableBinding(Processor.class) @Import(TestChannelBinderConfiguration.class) public static class StringMessageToStringMessageStreamListener { @@ -534,7 +534,7 @@ public class ContentTypeTckTests { return MessageBuilder.withPayload(person.toString()).setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN).build(); } } - + @EnableBinding(Processor.class) @Import(TestChannelBinderConfiguration.class) public static class ByteArrayMessageToStringJsonMessageStreamListener { @@ -547,7 +547,7 @@ public class ContentTypeTckTests { return MessageBuilder.withPayload(json).build(); } } - + @EnableBinding(Processor.class) @Import(TestChannelBinderConfiguration.class) public static class InternalPipeLine { @@ -556,19 +556,19 @@ public class ContentTypeTckTests { public String handleA(Person value) { return "{\"name\":\"" + value.getName().toUpperCase() + "\"}"; } - + @Bean public MessageChannel internalChannel() { return new DirectChannel(); } - + @StreamListener("internalChannel") @SendTo(Processor.OUTPUT) public String handleB(Person value) { return value.toString(); } } - + public static class Employee

{ private P person; private int id; @@ -585,14 +585,14 @@ public class ContentTypeTckTests { this.person = person; } } - + public static class Person { private String name; - + public Person() { this(null); } - + public Person(String name) { this.name = name; } @@ -604,26 +604,26 @@ public class ContentTypeTckTests { public void setName(String name) { this.name = name; } - + public String toString() { return name; } } - + @Configuration - public static class CustomConverters { + public static class CustomConverters { @Bean @StreamMessageConverter public FooBarMessageConverter fooBarMessageConverter() { return new FooBarMessageConverter(MimeType.valueOf("foo/bar")); } - + @Bean @StreamMessageConverter public AlwaysStringKryoMessageConverter kryoOverrideMessageConverter() { return new AlwaysStringKryoMessageConverter(MimeType.valueOf("application/x-java-object")); } - + /** * Even though this MessageConverter has nothing to do with Kryo it still shows how Kryo * conversion can be customized/overriden since it simply overriding a converter for @@ -634,12 +634,12 @@ public class ContentTypeTckTests { public AlwaysStringKryoMessageConverter(MimeType supportedMimeType) { super(supportedMimeType); } - + @Override protected boolean supports(Class clazz) { return clazz == null || String.class.isAssignableFrom(clazz); } - + protected Object convertFromInternal( Message message, Class targetClass, @Nullable Object conversionHint) { return this.getClass().getSimpleName(); @@ -649,7 +649,7 @@ public class ContentTypeTckTests { return ((String)payload).getBytes(StandardCharsets.UTF_8); } } - + public static class FooBarMessageConverter extends AbstractMessageConverter { protected FooBarMessageConverter(MimeType supportedMimeType) { super(supportedMimeType); @@ -657,13 +657,13 @@ public class ContentTypeTckTests { @Override protected boolean supports(Class clazz) { return clazz != null && String.class.isAssignableFrom(clazz); - } - + } + protected Object convertFromInternal( Message message, Class targetClass, @Nullable Object conversionHint) { return this.getClass().getSimpleName(); } - + protected Object convertToInternal( Object payload, @Nullable MessageHeaders headers, @Nullable Object conversionHint) { return ((String)payload).getBytes(StandardCharsets.UTF_8); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/CustomPartitionedProducerTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/CustomPartitionedProducerTest.java index 730f4df6a..ffbd9a4da 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/CustomPartitionedProducerTest.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/CustomPartitionedProducerTest.java @@ -80,7 +80,7 @@ public class CustomPartitionedProducerTest { } } } - + @Test public void testCustomPartitionedProducerByName() { ApplicationContext context = SpringApplication.run(CustomPartitionedProducerTest.TestSource.class, @@ -111,7 +111,7 @@ public class CustomPartitionedProducerTest { } } } - + @Test public void testCustomPartitionedProducerAsSingletons() { ApplicationContext context = SpringApplication.run(CustomPartitionedProducerTest.TestSource.class, @@ -139,7 +139,7 @@ public class CustomPartitionedProducerTest { } } } - + public void testCustomPartitionedProducerMultipleInstances() { ApplicationContext context = SpringApplication.run(CustomPartitionedProducerTest.TestSourceMultipleStrategies.class, "--spring.jmx.enabled=false", @@ -169,7 +169,7 @@ public class CustomPartitionedProducerTest { } } } - + @Test(expected=Exception.class) // It actually throws UnsatisfiedDependencyException, but it is confusing when it comes to test // But for the purposes of the test all we care about is that it fails @@ -188,12 +188,12 @@ public class CustomPartitionedProducerTest { public CustomPartitionSelectorClass customPartitionSelector() { return new CustomPartitionSelectorClass(); } - + @Bean public CustomPartitionKeyExtractorClass customPartitionKeyExtractor() { return new CustomPartitionKeyExtractorClass(); } - + @Bean @InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "5000", maxMessagesPerPoll = "1")) public MessageSource timerMessageSource() { @@ -205,7 +205,7 @@ public class CustomPartitionedProducerTest { }; } } - + @EnableBinding(Source.class) @EnableAutoConfiguration @Import(MockBinderRegistryConfiguration.class) @@ -216,22 +216,22 @@ public class CustomPartitionedProducerTest { public CustomPartitionSelectorClass customPartitionSelectorOne() { return new CustomPartitionSelectorClass(); } - + @Bean public CustomPartitionSelectorClass customPartitionSelectorTwo() { return new CustomPartitionSelectorClass(); } - + @Bean public CustomPartitionKeyExtractorClass customPartitionKeyExtractorOne() { return new CustomPartitionKeyExtractorClass(); } - + @Bean public CustomPartitionKeyExtractorClass customPartitionKeyExtractorTwo() { return new CustomPartitionKeyExtractorClass(); } - + @Bean @InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "5000", maxMessagesPerPoll = "1")) public MessageSource timerMessageSource() { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BinderPropertiesTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BinderPropertiesTests.java index 058a7caf6..d524f2149 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BinderPropertiesTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BinderPropertiesTests.java @@ -61,9 +61,9 @@ public class BinderPropertiesTests { ConfigurationPropertiesReportEndpoint endpoint = new ConfigurationPropertiesReportEndpoint(); endpoint.setApplicationContext(context); - + ContextConfigurationProperties configurationProperties = endpoint.configurationProperties().getContexts().values().iterator().next(); - + Map properties = configurationProperties.getBeans().get("bindingServiceProperties").getProperties(); assertFalse(properties.containsKey("error")); assertTrue(properties.containsKey("binders")); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BindingServiceConfigurationTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BindingServiceConfigurationTests.java index 795002c14..fca595163 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BindingServiceConfigurationTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BindingServiceConfigurationTests.java @@ -46,7 +46,7 @@ public class BindingServiceConfigurationTests { assertTrue(handler.getNotPropagatedHeaders().contains("contentType")); } } - + @Configuration @Import(ImportedConfiguration.class) public static class RootConfiguration { @@ -54,13 +54,13 @@ public class BindingServiceConfigurationTests { public void rootService(String val) { } } - + @Configuration public static class ImportedConfiguration { @ServiceActivator(inputChannel="input") public void importedService(String val) { } } - - + + }