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

Fix more trailing whitespace (tabs)

Browse Source
This commit is contained in:
Gary Russell
2018-03-12 10:26:46 -04:00
parent e1fc583d1e
commit 12bd608908
28 changed files with 171 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -85,9 +85,9 @@ public abstract class AbstractBinderTests<B extends AbstractTestBinder<? extends
protected final Log logger = LogFactory.getLog(this.getClass());
protected B testBinder;
protected SmartMessageConverter messageConverter;
@Before
public void before() {
this.messageConverter = new CompositeMessageConverterFactory().getMessageConverterForAllRegistered();
@@ -274,7 +274,7 @@ public abstract class AbstractBinderTests<B extends AbstractTestBinder<? extends
public void testSendAndReceiveJavaSerialization() throws Exception {
Binder binder = getBinder();
BindingProperties outputBindingProperties = createProducerBindingProperties(createProducerProperties());
DirectChannel moduleOutputChannel = createBindableChannel("output", outputBindingProperties);
BindingProperties inputBindingProperties = createConsumerBindingProperties(createConsumerProperties());
@@ -282,7 +282,7 @@ public abstract class AbstractBinderTests<B extends AbstractTestBinder<? extends
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());

10
spring-cloud-stream-core-docs/src/main/asciidoc/preface.adoc
View File

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

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

@@ -228,7 +228,7 @@ public class ContentTypeTests {
assertThat(message.getPayload()).isEqualTo(user.toString());
}
}
@Test
public void testSendTuple() throws Exception {
try (ConfigurableApplicationContext context = SpringApplication.run(

2
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroMessageConverterAutoConfiguration.java
View File

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

4
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/avro/AvroSchemaRegistryClientMessageConverter.java
View File

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

4
spring-cloud-stream-schema/src/main/java/org/springframework/cloud/stream/schema/client/DefaultSchemaRegistryClient.java
View File

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

16
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java
View File

@@ -36,11 +36,11 @@ import org.springframework.integration.endpoint.Pausable;
* @see org.springframework.cloud.stream.annotation.EnableBinding
*/
public interface Binding<T> extends Pausable {
default Map<String, Object> 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<T> 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<T> 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<T> 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<T> 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<T> extends Pausable {
default String getName() {
return null;
}
/**
* Unbinds the target component represented by this instance and stops any active

4
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java
View File

@@ -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<String, String> binderProperties = new HashMap<>();
this.flatten(null, binderConfiguration.getProperties(), binderProperties);
@@ -214,7 +214,7 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl
}
return (Binder<T, ?, ?>) this.binderInstanceCache.get(configurationName).getKey();
}
/**
* Ensures that nested properties are flattened (i.e., foo.bar=baz instead of foo={bar=baz})
*/

26
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java
View File

@@ -42,7 +42,7 @@ import org.springframework.util.StringUtils;
@JsonPropertyOrder({ "name", "group", "pausable", "state"})
@JsonIgnoreProperties("running")
public class DefaultBinding<T> implements Binding<T> {
private final Log logger = LogFactory.getLog(this.getClass().getName());
protected final String name;
@@ -52,9 +52,9 @@ public class DefaultBinding<T> implements Binding<T> {
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<T> implements Binding<T> {
this.target = target;
this.lifecycle = lifecycle;
}
public String getName() {
return this.name;
}
@@ -81,7 +81,7 @@ public class DefaultBinding<T> implements Binding<T> {
public String getGroup() {
return this.group;
}
public String getState() {
String state = "N/A";
if (this.lifecycle != null) {
@@ -94,15 +94,15 @@ public class DefaultBinding<T> implements Binding<T> {
}
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<T> implements Binding<T> {
}
}
}
@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<T> implements Binding<T> {
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<T> implements Binding<T> {
: 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";
}

10
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java
View File

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

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

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

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

@@ -217,7 +217,7 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Obje
public void bindInputs(BindingService bindingService) {
this.createAndBindInputs(bindingService);
}
@Override
public Collection<Binding<Object>> createAndBindInputs(BindingService bindingService) {
List<Binding<Object>> bindings = new ArrayList<>();

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

@@ -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<? extends MessageChannel> bindingTargetFactory,
DynamicDestinationsBindable dynamicDestinationsBindable) {
this(bindingService, bindingTargetFactory, dynamicDestinationsBindable, null, null);
}
@SuppressWarnings("rawtypes")
public BinderAwareChannelResolver(BindingService bindingService,
AbstractBindingTargetFactory<? extends MessageChannel> 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<MessageChannel> 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.");

2
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/InputBindingLifecycle.java
View File

@@ -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<Binding<Object>> inputBindings;

2
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageSourceBindingTargetFactory.java
View File

@@ -31,7 +31,7 @@ public class MessageSourceBindingTargetFactory
extends AbstractBindingTargetFactory<PollableMessageSource> {
private final MessageChannelAndSourceConfigurer messageSourceConfigurer;
private final SmartMessageConverter messageConverter;
public MessageSourceBindingTargetFactory(SmartMessageConverter messageConverter, MessageChannelConfigurer messageSourceConfigurer) {

4
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java
View File

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

4
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java
View File

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

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

@@ -39,7 +39,7 @@ import org.springframework.context.annotation.Configuration;
@AutoConfigureAfter(EndpointAutoConfiguration.class)
public class BindingsEndpointAutoConfiguration {
@Bean
@Bean
public BindingsEndpoint bindingsEndpoint(List<InputBindingLifecycle> inputBindings) {
return new BindingsEndpoint(inputBindings);
}

14
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java
View File

@@ -45,15 +45,15 @@ import org.springframework.messaging.converter.MessageConversionException;
*
*/
class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageConverter {
private final Map<ParameterizedTypeReference<?>, 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();

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

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

22
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/endpoint/BindingsEndpoint.java
View File

@@ -42,17 +42,17 @@ import org.springframework.util.Assert;
*/
@Endpoint(id = "bindings")
public class BindingsEndpoint {
private final List<InputBindingLifecycle> inputBindingLifecycles;
private final ObjectMapper objectMapper;
public BindingsEndpoint(List<InputBindingLifecycle> 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<Binding<?>> gatherInputBindings() {
List<Binding<?>> 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,

10
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/ApplicationMetricsProperties.java
View File

@@ -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<Map<String, String>> 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;
}
}

2
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/MetricsPublisherConfig.java
View File

@@ -26,7 +26,7 @@ import io.micrometer.core.instrument.step.StepRegistryConfig;
*
*/
class MetricsPublisherConfig implements StepRegistryConfig {
private final ApplicationMetricsProperties applicationMetricsProperties;
MetricsPublisherConfig(ApplicationMetricsProperties applicationMetricsProperties) {

4
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java
View File

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

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

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

22
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/CustomPartitionedProducerTest.java
View File

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

4
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BinderPropertiesTests.java
View File

@@ -61,9 +61,9 @@ public class BinderPropertiesTests {
ConfigurationPropertiesReportEndpoint endpoint = new ConfigurationPropertiesReportEndpoint();
endpoint.setApplicationContext(context);
ContextConfigurationProperties configurationProperties = endpoint.configurationProperties().getContexts().values().iterator().next();
Map<String, Object> properties = configurationProperties.getBeans().get("bindingServiceProperties").getProperties();
assertFalse(properties.containsKey("error"));
assertTrue(properties.containsKey("binders"));

8
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BindingServiceConfigurationTests.java
View File

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