diff --git a/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc b/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc index da07ea7ab..130cbf04c 100644 --- a/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc +++ b/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 payload = (Map) received.getPayload() + ... + + }, new ParameterizedTypeReference>() {})) +---- + ==== Reactive Programming Support Spring Cloud Stream also supports the use of reactive APIs where incoming and outgoing data is handled as continuous data flows. diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java index 0bae275ba..09a25fc7e 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java @@ -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> 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); } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PollableSource.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PollableSource.java index db3226e67..9faf8149f 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PollableSource.java +++ b/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 { /** @@ -34,4 +37,14 @@ public interface PollableSource { */ 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); + } + } 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 new file mode 100644 index 000000000..c188e2e37 --- /dev/null +++ b/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, 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 aa0ddfc4d..0052bd92e 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 @@ -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 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()); + }, new ParameterizedTypeReference() {})).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() {})).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 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()); + }, new ParameterizedTypeReference>() {})).isTrue(); + @SuppressWarnings("unchecked") + List list = (List) 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 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()); + }, new ParameterizedTypeReference>() {})).isTrue(); + @SuppressWarnings("unchecked") + Map map = (Map) 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; + } + + } + }