Pollable Consumer Payload Conversion

- support payload conversion for polled consumers

Cache Types

spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc

@@ -658,6 +658,24 @@ IMPORTANT: You must ack (or nack) the message at some point, to avoid resource l
 
 IMPORTANT: Some messaging systems (such as Apache Kafka) maintain a simple offset in a log, if a delivery fails and is requeued with  `StaticMessageHeaderAccessor.getAcknowledgmentCallback(m).acknowledge(Status.REQUEUE);`, any later successfully ack'd messages will be redelivered.
 
+There is also an overloaded `poll` method:
+
+[source,java]
+----
+poll(MessageHandler handler, ParameterizedTypeReference<?> type)
+----
+
+The `type` is a conversion hint allowing the incoming message payload to be converted:
+
+[source,java]
+----
+boolean result = pollableSource.poll(received -> {
+			Map<String, Foo> payload = (Map<String, Foo>) received.getPayload()
+            ...
+
+		}, new ParameterizedTypeReference<Map<String, Foo>>() {}))
+----
+
 ==== Reactive Programming Support
 
 Spring Cloud Stream also supports the use of reactive APIs where incoming and outgoing data is handled as continuous data flows.

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

@@ -27,6 +27,7 @@ import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.aop.support.NameMatchMethodPointcutAdvisor;
 import org.springframework.context.Lifecycle;
 import org.springframework.core.AttributeAccessor;
+import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.support.AckUtils;
@@ -39,7 +40,9 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessageHandlingException;
+import org.springframework.messaging.converter.SmartMessageConverter;
 import org.springframework.messaging.support.ChannelInterceptor;
+import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.retry.RecoveryCallback;
 import org.springframework.retry.RetryCallback;
 import org.springframework.retry.RetryContext;
@@ -74,7 +77,9 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life
 
 	private BiConsumer<AttributeAccessor, Message<?>> attributesProvider;
 
-	private volatile boolean running;
+	private SmartMessageConverter messageConverter;
+
+	private boolean running;
 
 	public void setSource(MessageSource<?> source) {
 		ProxyFactory pf = new ProxyFactory(source);
@@ -129,6 +134,10 @@ 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);
 	}
@@ -138,12 +147,12 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life
 	}
 
 	@Override
-	public boolean isRunning() {
+	public synchronized boolean isRunning() {
 		return this.running;
 	}
 
 	@Override
-	public void start() {
+	public synchronized void start() {
 		if (!this.running && this.source instanceof Lifecycle) {
 			((Lifecycle) this.source).start();
 		}
@@ -151,7 +160,7 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life
 	}
 
 	@Override
-	public void stop() {
+	public synchronized void stop() {
 		if (this.running && this.source instanceof Lifecycle) {
 			((Lifecycle) this.source).stop();
 		}
@@ -184,10 +193,21 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life
 
 	@Override
 	public boolean poll(MessageHandler handler) {
+		return poll(handler, null);
+	}
+
+	@Override
+	public boolean poll(MessageHandler handler, ParameterizedTypeReference<?> type) {
 		Message<?> message = this.source.receive();
 		if (message == null) {
 			return false;
 		}
+		if (type != null && this.messageConverter != null) {
+			Object payload = this.messageConverter.fromMessage(message, byte[].class, type);
+			message = MessageBuilder.withPayload(payload)
+					.copyHeaders(message.getHeaders())
+					.build();
+		}
 		AcknowledgmentCallback ackCallback = StaticMessageHeaderAccessor
 				.getAcknowledgmentCallback(message);
 		try {
@@ -211,9 +231,10 @@ public class DefaultPollableMessageSource implements PollableMessageSource, Life
 				}
 			}
 			else {
+				final Message<?> messageToHandle = message;
 				this.retryTemplate.execute(context -> {
-					setAttributesIfNecessary(message);
-					doHandleMessage(handler, message);
+					setAttributesIfNecessary(messageToHandle);
+					doHandleMessage(handler, messageToHandle);
 					return null;
 				}, this.recoveryCallback);
 			}

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

@@ -16,6 +16,8 @@
 
 package org.springframework.cloud.stream.binder;
 
+import org.springframework.core.ParameterizedTypeReference;
+
 /**
  * A mechanism to poll a consumer.
  *
@@ -25,6 +27,7 @@ package org.springframework.cloud.stream.binder;
  * @since 2.0
  *
  */
+@FunctionalInterface
 public interface PollableSource<H> {
 
 	/**
@@ -34,4 +37,14 @@ public interface PollableSource<H> {
 	 */
 	boolean poll(H handler);
 
+	/**
+	 * Poll the consumer and convert the payload to the type.
+	 * @param handler the handler.
+	 * @param type the type.
+	 * @return true if a message was handled.
+	 */
+	default boolean poll(H handler, ParameterizedTypeReference<?> type) {
+		return poll(handler);
+	}
+
 }

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

@@ -0,0 +1,92 @@
+/*
+ * 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

@@ -17,14 +17,19 @@
 package org.springframework.cloud.stream.binder;
 
 import java.util.Collections;
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.Test;
 
 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.core.ParameterizedTypeReference;
 import org.springframework.integration.channel.PublishSubscribeChannel;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.messaging.Message;
@@ -76,6 +81,71 @@ public class PollableConsumerTests {
 		assertThat(count.get()).isEqualTo(2);
 	}
 
+	@Test
+	public void testConvertSimple() {
+		SpringIntegrationChannelBinder binder = createBinder();
+		binder.setMessageSourceDelegate(() -> new GenericMessage<>("{\"foo\":\"bar\"}"));
+		DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource();
+		pollableSource.setMessageConverter(new SmartJsonMessageConverter());
+		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());
+		}, new ParameterizedTypeReference<Foo>() {})).isTrue();
+		assertThat(payload.get()).isInstanceOf(Foo.class);
+		assertThat(((Foo) payload.get()).getFoo()).isEqualTo("bar");
+		// test the cache for coverage
+		assertThat(pollableSource.poll(received -> {
+			payload.set(received.getPayload());
+		}, new ParameterizedTypeReference<Foo>() {})).isTrue();
+		assertThat(payload.get()).isInstanceOf(Foo.class);
+		assertThat(((Foo) payload.get()).getFoo()).isEqualTo("bar");
+	}
+
+	@Test
+	public void testConvertList() {
+		SpringIntegrationChannelBinder binder = createBinder();
+		binder.setMessageSourceDelegate(() -> new GenericMessage<>("[{\"foo\":\"bar\"},{\"foo\":\"baz\"}]"));
+		DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource();
+		pollableSource.setMessageConverter(new SmartJsonMessageConverter());
+		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());
+		}, new ParameterizedTypeReference<List<Foo>>() {})).isTrue();
+		@SuppressWarnings("unchecked")
+		List<Foo> list = (List<Foo>) payload.get();
+		assertThat(list.size()).isEqualTo(2);
+		assertThat(list.get(0).getFoo()).isEqualTo("bar");
+		assertThat(list.get(1).getFoo()).isEqualTo("baz");
+	}
+
+	@Test
+	public void testConvertMap() {
+		SpringIntegrationChannelBinder binder = createBinder();
+		binder.setMessageSourceDelegate(() -> new GenericMessage<>("{\"qux\":{\"foo\":\"bar\"}}"));
+		DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource();
+		pollableSource.setMessageConverter(new SmartJsonMessageConverter());
+		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());
+		}, new ParameterizedTypeReference<Map<String, Foo>>() {})).isTrue();
+		@SuppressWarnings("unchecked")
+		Map<String, Foo> map = (Map<String, Foo>) payload.get();
+		assertThat(map.size()).isEqualTo(1);
+		assertThat(map.get("qux").getFoo()).isEqualTo("bar");
+	}
+
 	@Test
 	public void testEmbedded() {
 		SpringIntegrationChannelBinder binder = createBinder();
@@ -182,4 +252,18 @@ public class PollableConsumerTests {
 		return binder;
 	}
 
+	public static class Foo {
+
+		private String foo;
+
+		protected String getFoo() {
+			return this.foo;
+		}
+
+		protected void setFoo(String foo) {
+			this.foo = foo;
+		}
+
+	}
+
 }