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 5867cceff..5c0d2d619 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 @@ -31,6 +31,7 @@ import java.util.concurrent.atomic.AtomicReference; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.junit.After; +import org.junit.Before; import org.junit.Test; import org.springframework.cloud.stream.binder.AbstractBinderTests.Station.Readings; @@ -56,6 +57,7 @@ import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.converter.SmartMessageConverter; import org.springframework.messaging.handler.annotation.support.PayloadArgumentResolver; import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite; import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; @@ -84,6 +86,13 @@ public abstract class AbstractBinderTests interceptors = new ArrayList<>(); private final MessagingTemplate messagingTemplate = new MessagingTemplate(); + + private final SmartMessageConverter messageConverter; private MessageSource source; @@ -77,9 +80,11 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life private BiConsumer> attributesProvider; - private SmartMessageConverter messageConverter; - private boolean running; + + public DefaultPollableMessageSource(SmartMessageConverter messageConverter) { + this.messageConverter = messageConverter; + } public void setSource(MessageSource source) { ProxyFactory pf = new ProxyFactory(source); @@ -134,10 +139,6 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life this.attributesProvider = attributesProvider; } - public void setMessageConverter(SmartMessageConverter messageConverter) { - this.messageConverter = messageConverter; - } - public void addInterceptor(ChannelInterceptor interceptor) { this.interceptors.add(interceptor); } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingBeanDefinitionRegistryUtils.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingBeanDefinitionRegistryUtils.java index 96be7ea66..f02ec4fda 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingBeanDefinitionRegistryUtils.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingBeanDefinitionRegistryUtils.java @@ -94,10 +94,6 @@ public abstract class BindingBeanDefinitionRegistryUtils { }); } - /** - * @deprecated as of version 2.0 based on deprecated {@link Bindings} annotation. - */ - @Deprecated public static void registerBindingTargetsQualifiedBeanDefinitions(Class parent, Class type, final BeanDefinitionRegistry registry) { 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 0f01e3977..53d386e11 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 @@ -18,6 +18,7 @@ package org.springframework.cloud.stream.binding; import org.springframework.cloud.stream.binder.DefaultPollableMessageSource; import org.springframework.cloud.stream.binder.PollableMessageSource; +import org.springframework.messaging.converter.SmartMessageConverter; import org.springframework.util.Assert; /** @@ -30,16 +31,19 @@ public class MessageSourceBindingTargetFactory extends AbstractBindingTargetFactory { private final MessageChannelAndSourceConfigurer messageSourceConfigurer; + + private final SmartMessageConverter messageConverter; - public MessageSourceBindingTargetFactory(MessageChannelConfigurer messageSourceConfigurer) { + public MessageSourceBindingTargetFactory(SmartMessageConverter messageConverter, MessageChannelConfigurer messageSourceConfigurer) { super(PollableMessageSource.class); Assert.isInstanceOf(MessageChannelAndSourceConfigurer.class, messageSourceConfigurer); this.messageSourceConfigurer = (MessageChannelAndSourceConfigurer) messageSourceConfigurer; + this.messageConverter = messageConverter; } @Override public PollableMessageSource createInput(String name) { - DefaultPollableMessageSource binding = new DefaultPollableMessageSource(); + DefaultPollableMessageSource binding = new DefaultPollableMessageSource(this.messageConverter); this.messageSourceConfigurer.configurePolledMessageSource(binding, name); return binding; } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingBeansRegistrar.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingBeansRegistrar.java index 705e3dfe2..f5b780e04 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingBeansRegistrar.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingBeansRegistrar.java @@ -17,7 +17,6 @@ package org.springframework.cloud.stream.config; import org.springframework.beans.factory.support.BeanDefinitionRegistry; -import org.springframework.cloud.stream.annotation.Bindings; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils; import org.springframework.context.annotation.ImportBeanDefinitionRegistrar; @@ -28,13 +27,11 @@ import org.springframework.core.type.AnnotationMetadata; import org.springframework.util.ClassUtils; /** - * @deprecated as ov version 2.0 based on deprecation of {@link Bindings} anntoation * * @author Marius Bogoevici * @author Dave Syer * @author Artem Bilan */ -@Deprecated public class BindingBeansRegistrar implements ImportBeanDefinitionRegistrar { @Override diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java index 805e915a9..80a928b8e 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java @@ -138,9 +138,9 @@ public class BindingServiceConfiguration { } @Bean - public MessageSourceBindingTargetFactory messageSourceFactory( + public MessageSourceBindingTargetFactory messageSourceFactory(CompositeMessageConverterFactory compositeMessageConverterFactory, CompositeMessageChannelConfigurer compositeMessageChannelConfigurer) { - return new MessageSourceBindingTargetFactory(compositeMessageChannelConfigurer); + return new MessageSourceBindingTargetFactory(compositeMessageConverterFactory.getMessageConverterForAllRegistered(), compositeMessageChannelConfigurer); } @Bean 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 d4aba23ea..1e0bf363a 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 @@ -16,13 +16,21 @@ package org.springframework.cloud.stream.converter; +import java.io.IOException; import java.nio.charset.StandardCharsets; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import com.fasterxml.jackson.databind.JavaType; +import com.fasterxml.jackson.databind.ObjectMapper; import org.springframework.core.MethodParameter; +import org.springframework.core.ParameterizedTypeReference; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.MappingJackson2MessageConverter; +import org.springframework.messaging.converter.MessageConversionException; /** * Variation of {@link MappingJackson2MessageConverter} to support marshalling and @@ -30,10 +38,14 @@ import org.springframework.messaging.converter.MappingJackson2MessageConverter; * * * @author Oleg Zhurakousky + * @author Gary Russell * @since 2.0 * */ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageConverter { + + private final Map, JavaType> typeCache = new ConcurrentHashMap<>(); + @Override protected Object convertToInternal(Object payload, @Nullable MessageHeaders headers, @Nullable Object conversionHint) { @@ -62,12 +74,44 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC conversionHint = null; } } - if (message.getPayload() instanceof byte[] && targetClass.isAssignableFrom(String.class)) { - result = new String((byte[])message.getPayload(), StandardCharsets.UTF_8); + else if (conversionHint instanceof ParameterizedTypeReference) { + result = convertParameterizedType(message, targetClass, conversionHint); } - else { - result = super.convertFromInternal(message, targetClass, conversionHint); + + if (result == null) { + if (message.getPayload() instanceof byte[] && targetClass.isAssignableFrom(String.class)) { + result = new String((byte[])message.getPayload(), StandardCharsets.UTF_8); + } + else { + result = super.convertFromInternal(message, targetClass, conversionHint); + } } + return result; } + + private Object convertParameterizedType(Message message, Class targetClass, Object conversionHint) { + ObjectMapper objectMapper = this.getObjectMapper(); + Object payload = message.getPayload(); + try { + JavaType type = this.typeCache.get(conversionHint); + if (type == null) { + type = objectMapper.getTypeFactory().constructType( + ((ParameterizedTypeReference) conversionHint).getType()); + this.typeCache.put((ParameterizedTypeReference) conversionHint, type); + } + if (payload instanceof byte[]) { + return objectMapper.readValue((byte[]) payload, type); + } + else if (payload instanceof String) { + return objectMapper.readValue((String) payload, type); + } + else { + return null; + } + } + catch (IOException e) { + throw new MessageConversionException("Cannot parse payload ", e); + } + } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/SmartJsonMessageConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/SmartJsonMessageConverter.java deleted file mode 100644 index c188e2e37..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/SmartJsonMessageConverter.java +++ /dev/null @@ -1,92 +0,0 @@ -/* - * Copyright 2018 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.stream.converter; - -import java.io.IOException; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; - -import com.fasterxml.jackson.databind.JavaType; -import com.fasterxml.jackson.databind.ObjectMapper; - -import org.springframework.core.ParameterizedTypeReference; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHeaders; -import org.springframework.messaging.converter.MessageConversionException; -import org.springframework.messaging.converter.SmartMessageConverter; - -/** - * @author Gary Russell - * @since 2.0 - * - */ -public class SmartJsonMessageConverter implements SmartMessageConverter { - - private final ObjectMapper objectMapper; - - private final Map, JavaType> typeCache = new ConcurrentHashMap<>(); - - public SmartJsonMessageConverter() { - this.objectMapper = new ObjectMapper(); - } - - @Override - public Object fromMessage(Message message, Class targetClass) { - throw new UnsupportedOperationException(); - } - - @Override - public Message toMessage(Object payload, MessageHeaders headers) { - throw new UnsupportedOperationException(); - } - - @Override - public Object fromMessage(Message message, Class targetClass, Object conversionHint) { - Object payload = message.getPayload(); - try { - if (conversionHint instanceof ParameterizedTypeReference) { - JavaType type = this.typeCache.get(conversionHint); - if (type == null) { - type = this.objectMapper.getTypeFactory().constructType( - ((ParameterizedTypeReference) conversionHint).getType()); - this.typeCache.put((ParameterizedTypeReference) conversionHint, type); - } - if (payload instanceof byte[]) { - return this.objectMapper.readValue((byte[]) payload, type); - } - else if (payload instanceof String) { - return this.objectMapper.readValue((String) payload, type); - } - else { - throw new IllegalArgumentException("Unsupported payload type"); - } - } - else { - throw new IllegalArgumentException("Must provide a ParamterizedTypeReference"); - } - } - catch (IOException e) { - throw new MessageConversionException("Cannot parse payload ", e); - } - } - - @Override - public Message toMessage(Object payload, MessageHeaders headers, Object conversionHint) { - throw new UnsupportedOperationException(); - } - -} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java index 0052bd92e..8462d06ce 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java @@ -15,7 +15,6 @@ */ package org.springframework.cloud.stream.binder; - import java.util.Collections; import java.util.List; import java.util.Map; @@ -23,39 +22,55 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import org.junit.Before; import org.junit.Test; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.stream.binder.integration.SpringIntegrationBinderConfiguration; import org.springframework.cloud.stream.binder.integration.SpringIntegrationChannelBinder; -import org.springframework.cloud.stream.binder.integration.SpringIntegrationProvisioner; -import org.springframework.cloud.stream.converter.SmartJsonMessageConverter; -import org.springframework.context.support.GenericApplicationContext; +import org.springframework.cloud.stream.binding.MessageConverterConfigurer; +import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; +import org.springframework.context.ApplicationContext; import org.springframework.core.ParameterizedTypeReference; -import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.SubscribableChannel; +import org.springframework.messaging.converter.SmartMessageConverter; import org.springframework.messaging.support.ChannelInterceptorAdapter; import org.springframework.messaging.support.GenericMessage; import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.MimeType; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.fail; /** * @author Gary Russell + * @author Oleg Zhurakousky * @since 2.0 * */ public class PollableConsumerTests { - private final GenericApplicationContext context = new GenericApplicationContext(); + private ApplicationContext context; + + private SmartMessageConverter messageConverter; + + @Before + public void before() { + this.messageConverter = new CompositeMessageConverterFactory().getMessageConverterForAllRegistered(); + } @Test public void testSimple() { SpringIntegrationChannelBinder binder = createBinder(); - DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(); + MessageConverterConfigurer configurer = context.getBean(MessageConverterConfigurer.class); + + DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(this.messageConverter); + configurer.configurePolledMessageSource(pollableSource, "foo"); pollableSource.addInterceptor(new ChannelInterceptorAdapter() { @Override @@ -73,7 +88,7 @@ public class PollableConsumerTests { final AtomicInteger count = new AtomicInteger(); assertThat(pollableSource.poll(received -> { assertThat(received.getPayload()).isEqualTo("POLLED DATA"); - assertThat(received.getHeaders().get(MessageHeaders.CONTENT_TYPE)).isEqualTo("text/plain"); + assertThat(received.getHeaders().get(MessageHeaders.CONTENT_TYPE)).isEqualTo(MimeType.valueOf("text/plain")); if (count.incrementAndGet() == 1) { throw new RuntimeException("test retry"); } @@ -84,9 +99,12 @@ public class PollableConsumerTests { @Test public void testConvertSimple() { SpringIntegrationChannelBinder binder = createBinder(); - binder.setMessageSourceDelegate(() -> new GenericMessage<>("{\"foo\":\"bar\"}")); - DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(); - pollableSource.setMessageConverter(new SmartJsonMessageConverter()); + MessageConverterConfigurer configurer = context.getBean(MessageConverterConfigurer.class); + + binder.setMessageSourceDelegate(() -> new GenericMessage<>("{\"foo\":\"bar\"}".getBytes())); + DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(this.messageConverter); + configurer.configurePolledMessageSource(pollableSource, "foo"); + ExtendedConsumerProperties properties = new ExtendedConsumerProperties<>(null); properties.setMaxAttempts(1); properties.setBackOffInitialInterval(0); @@ -108,13 +126,20 @@ public class PollableConsumerTests { @Test public void testConvertList() { SpringIntegrationChannelBinder binder = createBinder(); - binder.setMessageSourceDelegate(() -> new GenericMessage<>("[{\"foo\":\"bar\"},{\"foo\":\"baz\"}]")); - DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(); - pollableSource.setMessageConverter(new SmartJsonMessageConverter()); + MessageConverterConfigurer configurer = context.getBean(MessageConverterConfigurer.class); + + binder.setMessageSourceDelegate(() -> new GenericMessage<>("[{\"foo\":\"bar\"},{\"foo\":\"baz\"}]".getBytes())); + DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(this.messageConverter); + configurer.configurePolledMessageSource(pollableSource, "foo"); + ExtendedConsumerProperties properties = new ExtendedConsumerProperties<>(null); properties.setMaxAttempts(1); properties.setBackOffInitialInterval(0); + + + binder.bindPollableConsumer("foo", "bar", pollableSource, properties); + final AtomicReference payload = new AtomicReference<>(); assertThat(pollableSource.poll(received -> { payload.set(received.getPayload()); @@ -129,9 +154,12 @@ public class PollableConsumerTests { @Test public void testConvertMap() { SpringIntegrationChannelBinder binder = createBinder(); - binder.setMessageSourceDelegate(() -> new GenericMessage<>("{\"qux\":{\"foo\":\"bar\"}}")); - DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(); - pollableSource.setMessageConverter(new SmartJsonMessageConverter()); + MessageConverterConfigurer configurer = context.getBean(MessageConverterConfigurer.class); + + binder.setMessageSourceDelegate(() -> new GenericMessage<>("{\"qux\":{\"foo\":\"bar\"}}".getBytes())); + DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(this.messageConverter); + configurer.configurePolledMessageSource(pollableSource, "foo"); + ExtendedConsumerProperties properties = new ExtendedConsumerProperties<>(null); properties.setMaxAttempts(1); properties.setBackOffInitialInterval(0); @@ -149,6 +177,8 @@ public class PollableConsumerTests { @Test public void testEmbedded() { SpringIntegrationChannelBinder binder = createBinder(); + MessageConverterConfigurer configurer = context.getBean(MessageConverterConfigurer.class); + binder.setMessageSourceDelegate(() -> { MessageValues original = new MessageValues("foo".getBytes(), Collections.singletonMap(MessageHeaders.CONTENT_TYPE, "application/octet-stream")); @@ -163,7 +193,8 @@ public class PollableConsumerTests { }); ExtendedConsumerProperties properties = new ExtendedConsumerProperties<>(null); properties.setHeaderMode(HeaderMode.embeddedHeaders); - DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(); + DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(this.messageConverter); + configurer.configurePolledMessageSource(pollableSource, "foo"); pollableSource.addInterceptor(new ChannelInterceptorAdapter() { @Override @@ -184,7 +215,10 @@ public class PollableConsumerTests { @Test public void testErrors() { SpringIntegrationChannelBinder binder = createBinder(); - DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(); + MessageConverterConfigurer configurer = context.getBean(MessageConverterConfigurer.class); + + DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(this.messageConverter); + configurer.configurePolledMessageSource(pollableSource, "foo"); pollableSource.addInterceptor(new ChannelInterceptorAdapter() { @Override @@ -200,7 +234,7 @@ public class PollableConsumerTests { properties.setBackOffInitialInterval(0); binder.bindPollableConsumer("foo", "bar", pollableSource, properties); final CountDownLatch latch = new CountDownLatch(1); - this.context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, SubscribableChannel.class).subscribe(m -> { + context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, SubscribableChannel.class).subscribe(m -> { latch.countDown(); }); final AtomicInteger count = new AtomicInteger(); @@ -217,7 +251,10 @@ public class PollableConsumerTests { @Test public void testErrorsNoRetry() { SpringIntegrationChannelBinder binder = createBinder(); - DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(); + MessageConverterConfigurer configurer = context.getBean(MessageConverterConfigurer.class); + + DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource(this.messageConverter); + configurer.configurePolledMessageSource(pollableSource, "foo"); pollableSource.addInterceptor(new ChannelInterceptorAdapter() { @Override @@ -232,7 +269,7 @@ public class PollableConsumerTests { properties.setMaxAttempts(1); binder.bindPollableConsumer("foo", "bar", pollableSource, properties); final CountDownLatch latch = new CountDownLatch(1); - this.context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, SubscribableChannel.class).subscribe(m -> { + context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, SubscribableChannel.class).subscribe(m -> { latch.countDown(); }); final AtomicInteger count = new AtomicInteger(); @@ -244,11 +281,9 @@ public class PollableConsumerTests { } private SpringIntegrationChannelBinder createBinder() { - SpringIntegrationProvisioner provisioningProvider = new SpringIntegrationProvisioner(); - SpringIntegrationChannelBinder binder = new SpringIntegrationChannelBinder(provisioningProvider); - this.context.registerBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, PublishSubscribeChannel.class); - this.context.refresh(); - binder.setApplicationContext(this.context); + this.context = new SpringApplicationBuilder(SpringIntegrationBinderConfiguration.getCompleteConfiguration()) + .web(WebApplicationType.NONE).run(); + SpringIntegrationChannelBinder binder = context.getBean(SpringIntegrationChannelBinder.class); return binder; } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationChannelBinder.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationChannelBinder.java index fb32f5cf5..a5d1c210f 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationChannelBinder.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationChannelBinder.java @@ -120,7 +120,7 @@ public class SpringIntegrationChannelBinder extends AbstractMessageChannelBinder * Set a delegate {@link MessageSource} for pollable consumers. * @param messageSourceDelegate the delegate. */ - public void setMessageSourceDelegate(MessageSource messageSourceDelegate) { + public void setMessageSourceDelegate(MessageSource messageSourceDelegate) { this.messageSourceDelegate = messageSourceDelegate; }