From 5e8e3468afe37e82df035409c2aeb9f3ba25c42c Mon Sep 17 00:00:00 2001
From: Gary Russell <grussell@pivotal.io>
Date: Thu, 6 Sep 2018 12:57:30 -0400
Subject: [PATCH] GH-1459: Pollable Consumer and Requeue

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/1459

Requires https://github.com/spring-cloud/spring-cloud-stream/pull/1467
---
 .../binder/rabbit/RabbitBinderTests.java      | 40 +++++++++++++++++--
 1 file changed, 36 insertions(+), 4 deletions(-)

diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
index 5629bb726..b2f2f1eb7 100644
--- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
+++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
@@ -69,6 +69,7 @@ import org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy;
 import org.springframework.cloud.stream.binder.PartitionSelectorStrategy;
 import org.springframework.cloud.stream.binder.PartitionTestSupport;
 import org.springframework.cloud.stream.binder.PollableSource;
+import org.springframework.cloud.stream.binder.RequeueCurrentMessageException;
 import org.springframework.cloud.stream.binder.Spy;
 import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties;
 import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties;
@@ -95,7 +96,7 @@ import org.springframework.messaging.MessageHandlingException;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.SubscribableChannel;
-import org.springframework.messaging.support.ChannelInterceptorAdapter;
+import org.springframework.messaging.support.ChannelInterceptor;
 import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.retry.support.RetryTemplate;
@@ -1306,7 +1307,7 @@ public class RabbitBinderTests extends
 		admin.declareQueue(queue);
 		admin.declareBinding(binding);
 
-		output.addInterceptor(new ChannelInterceptorAdapter() {
+		output.addInterceptor(new ChannelInterceptor() {
 
 			@Override
 			public Message<?> preSend(Message<?> message, MessageChannel channel) {
@@ -1348,7 +1349,7 @@ public class RabbitBinderTests extends
 		admin.declareQueue(queue);
 		admin.declareBinding(binding);
 
-		output.addInterceptor(new ChannelInterceptorAdapter() {
+		output.addInterceptor(new ChannelInterceptor() {
 
 			@Override
 			public Message<?> preSend(Message<?> message, MessageChannel channel) {
@@ -1391,6 +1392,36 @@ public class RabbitBinderTests extends
 		binding.unbind();
 	}
 
+	@Test
+	public void testPolledConsumerRequeue() throws Exception {
+		RabbitTestBinder binder = getBinder();
+		PollableSource<MessageHandler> inboundBindTarget = new DefaultPollableMessageSource(this.messageConverter);
+		ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
+		Binding<PollableSource<MessageHandler>> binding = binder.bindPollableConsumer("pollableRequeue", "group",
+				inboundBindTarget, properties);
+		RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
+		template.convertAndSend("pollableRequeue.group", "testPollable");
+		try {
+			boolean polled = false;
+			int n = 0;
+			while (n++ < 100 && !polled) {
+				polled = inboundBindTarget.poll(m -> {
+					assertThat(m.getPayload()).isEqualTo("testPollable");
+					throw new RequeueCurrentMessageException();
+				});
+			}
+			fail("Expected exception");
+		}
+		catch (MessageHandlingException e) {
+			assertThat(e.getCause()).isInstanceOf(RequeueCurrentMessageException.class);
+		}
+		boolean polled = inboundBindTarget.poll(m -> {
+			assertThat(m.getPayload()).isEqualTo("testPollable");
+		});
+		assertThat(polled).isTrue();
+		binding.unbind();
+	}
+
 	@Test
 	public void testPolledConsumerWithDlq() throws Exception {
 		RabbitTestBinder binder = getBinder();
@@ -1413,7 +1444,8 @@ public class RabbitBinderTests extends
 			}
 		}
 		catch (MessageHandlingException e) {
-			assertThat(e.getCause().getCause().getCause().getCause().getCause().getMessage()).isEqualTo("test DLQ");
+			assertThat(e.getCause().getCause().getCause().getCause().getCause().getMessage())
+					.isEqualTo("test DLQ");
 		}
 		org.springframework.amqp.core.Message deadLetter = template.receive("pollableDlq.group.dlq", 10_000);
 		assertThat(deadLetter).isNotNull();