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 9d66b1601..cd3a35c80 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 @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2020 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. @@ -22,6 +22,8 @@ import java.util.function.BiConsumer; import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.springframework.aop.framework.ProxyFactory; import org.springframework.aop.support.NameMatchMethodPointcutAdvisor; @@ -59,12 +61,15 @@ import org.springframework.util.Assert; * * @author Gary Russell * @author Oleg Zhurakousky + * @author David Turanski * @since 2.0 * */ public class DefaultPollableMessageSource implements PollableMessageSource, Lifecycle, RetryListener { + private static final Log log = LogFactory.getLog(DefaultPollableMessageSource.class); + protected static final ThreadLocal attributesHolder = new ThreadLocal(); private static final DirectChannel dummyChannel = new DirectChannel(); @@ -204,6 +209,11 @@ public class DefaultPollableMessageSource AcknowledgmentCallback ackCallback = StaticMessageHeaderAccessor .getAcknowledgmentCallback(message); + + if (ackCallback == null) { + ackCallback = status -> log.warn("No AcknowledgementCallback defined. Status: " + status.name() + " " + message); + } + try { if (this.retryTemplate == null) { this.handle(message, handler); 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 b0a6df60b..d1685216f 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 @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2020 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. @@ -62,6 +62,7 @@ import static org.mockito.Mockito.verify; /** * @author Gary Russell * @author Oleg Zhurakousky + * @author David Turanski * @since 2.0 * */ @@ -77,6 +78,28 @@ public class PollableConsumerTests { .getMessageConverterForAllRegistered(); } + @Test + public void testDefaultMessageSource() { + TestChannelBinder binder = createBinder(); + MessageConverterConfigurer configurer = this.context + .getBean(MessageConverterConfigurer.class); + + DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource( + this.messageConverter); + configurer.configurePolledMessageSource(pollableSource, "foo"); + ExtendedConsumerProperties properties = new ExtendedConsumerProperties<>( + null); + properties.setMaxAttempts(2); + properties.setBackOffInitialInterval(0); + binder.bindPollableConsumer("foo", "bar", pollableSource, properties); + AtomicInteger count = new AtomicInteger(); + assertThat(pollableSource.poll(message -> { + assertThat(message).isNotNull(); + count.incrementAndGet(); + })).isTrue(); + assertThat(count.get()).isOne(); + } + @Test public void testSimple() { TestChannelBinder binder = createBinder(); @@ -414,6 +437,39 @@ public class PollableConsumerTests { verify(callback).acknowledge(Status.REQUEUE); } + @Test + public void testRequeueWithNoAcknowledgementCallback() { + TestChannelBinder binder = createBinder(); + MessageConverterConfigurer configurer = this.context + .getBean(MessageConverterConfigurer.class); + + DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource( + this.messageConverter); + configurer.configurePolledMessageSource(pollableSource, "foo"); + pollableSource.addInterceptor(new ChannelInterceptor() { + + @Override + public Message preSend(Message message, MessageChannel channel) { + return MessageBuilder.fromMessage(message) + .build(); + } + + }); + ExtendedConsumerProperties properties = new ExtendedConsumerProperties<>(null); + properties.setMaxAttempts(2); + properties.setBackOffInitialInterval(0); + binder.bindPollableConsumer("foo", "bar", pollableSource, properties); + final AtomicInteger count = new AtomicInteger(); + + assertThat(pollableSource.poll(received -> { + count.incrementAndGet(); + throw new RequeueCurrentMessageException("test retry"); + })).isTrue(); + + assertThat(count.get()).isEqualTo(2); + + } + @Test public void testRequeueFromErrorFlow() { TestChannelBinder binder = createBinder(); @@ -482,19 +538,19 @@ public class PollableConsumerTests { public void testAutoStartupOn() { TestChannelBinder binder = createBinder(); binder.setMessageSourceDelegate(new LifecycleMessageSource( - () -> new GenericMessage<>("{\"foo\":\"bar\"}".getBytes()))); + () -> new GenericMessage<>("{\"foo\":\"bar\"}".getBytes()))); MessageConverterConfigurer configurer = this.context - .getBean(MessageConverterConfigurer.class); + .getBean(MessageConverterConfigurer.class); DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource( - this.messageConverter); + this.messageConverter); configurer.configurePolledMessageSource(pollableSource, "foo"); ExtendedConsumerProperties properties = new ExtendedConsumerProperties<>( - null); + null); properties.setAutoStartup(true); Binding> pollableSourceBinding = binder - .bindPollableConsumer("foo", "bar", pollableSource, properties); + .bindPollableConsumer("foo", "bar", pollableSource, properties); assertThat(pollableSourceBinding.isRunning()).isTrue(); } @@ -523,6 +579,7 @@ public class PollableConsumerTests { public static class LifecycleMessageSource implements MessageSource, Lifecycle { private final MessageSource delegate; + private boolean running = false; public LifecycleMessageSource(MessageSource delegate) { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/TestChannelBinder.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/TestChannelBinder.java index 956a573cf..7cce8644b 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/TestChannelBinder.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/TestChannelBinder.java @@ -16,7 +16,6 @@ package org.springframework.cloud.stream.binder.test; -import java.util.Collections; import java.util.function.Consumer; import org.springframework.beans.factory.BeanFactory; @@ -30,12 +29,15 @@ import org.springframework.cloud.stream.binder.test.TestChannelBinderProvisioner import org.springframework.cloud.stream.provisioning.ConsumerDestination; import org.springframework.cloud.stream.provisioning.ProducerDestination; import org.springframework.core.AttributeAccessor; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.acks.AcknowledgmentCallback; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.core.MessageSource; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.handler.BridgeHandler; import org.springframework.integration.support.DefaultErrorMessageStrategy; import org.springframework.integration.support.ErrorMessageStrategy; +import org.springframework.integration.support.MapBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -110,8 +112,10 @@ public class TestChannelBinder extends private Message lastError; private MessageSource messageSourceDelegate = () -> new GenericMessage<>( - "polled data", - Collections.singletonMap(MessageHeaders.CONTENT_TYPE, "text/plain")); + "polled data", new MapBuilder() + .put(MessageHeaders.CONTENT_TYPE, "text/plain") + .put(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, (AcknowledgmentCallback) status -> { + }).get()); public TestChannelBinder(TestChannelBinderProvisioner provisioningProvider) { super(new String[] {}, provisioningProvider);