Added support for framework provided converter to DefaultPollableMessageSource

- Merged SmartJsonMessageConverter with ApplicationJsonMessageMarshallingConverter
- Fixed Tests
- Fixed SpringIntegrationChannelBinder to enforce byte[] payload type for polling delegate
- Other minor polishings

Resolves #1197

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<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();
+	}
 
 	/**
 	 * Subclasses may override this default value to have tests wait longer for a message
@@ -459,7 +468,7 @@ public abstract class AbstractBinderTests<B extends AbstractTestBinder<? extends
 
 	protected DefaultPollableMessageSource createBindableMessageSource(String bindingName,
 			BindingProperties bindingProperties) throws Exception {
-		DefaultPollableMessageSource source = new DefaultPollableMessageSource();
+		DefaultPollableMessageSource source = new DefaultPollableMessageSource(new CompositeMessageConverterFactory().getMessageConverterForAllRegistered());
 		createConverterConfigurer(bindingName, bindingProperties).configurePolledMessageSource(source, bindingName);
 		return source;
 	}

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

@@ -54,6 +54,7 @@ import org.springframework.util.Assert;
  * The default implementation of a {@link PollableMessageSource}.
  *
  * @author Gary Russell
+ * @author Oleg Zhurakousky
  * @since 2.0
  *
  */
@@ -64,6 +65,8 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life
 	private final List<ChannelInterceptor> 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<AttributeAccessor, Message<?>> 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);
 	}

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

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<PollableMessageSource> {
 
 	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;
 	}

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

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

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<ParameterizedTypeReference<?>, 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);
+		}
+	}
 }

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/SmartJsonMessageConverter.java

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

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<Object> 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<Object> properties = new ExtendedConsumerProperties<>(null);
 		properties.setMaxAttempts(1);
 		properties.setBackOffInitialInterval(0);
+		
+		
+		
 		binder.bindPollableConsumer("foo", "bar", pollableSource, properties);
+		
 		final AtomicReference<Object> 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<Object> 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<Object> 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;
 	}
 

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<byte[]> messageSourceDelegate) {
 		this.messageSourceDelegate = messageSourceDelegate;
 	}