From 126c53e8ad0743ee5ba21ae39da9d073f50b658f Mon Sep 17 00:00:00 2001
From: Thomas Recloux <thomas@recloux.fr>
Date: Wed, 10 Mar 2021 19:09:12 +0100
Subject: [PATCH] GH-151: Remove output bean from the Rabbit Supplier

Fixes https://github.com/spring-cloud/stream-applications/issues/151

The `RabbitSupplierConfiguration` comes with the `output` bean definition which
clashes with the `output` binding for Spring Cloud Stream.
The last one doesn't register the binding bean and leave the rest of the logic
to deal with existing bean when we resolve it by the `output` name.

* Rework the logic of the `RabbitSupplierConfiguration` to avoid an extra bean
in between and the binding in Spring Cloud Stream do it job
---
 .../rabbit/RabbitSourceListenerTests.java     |  1 +
 .../rabbit/RabbitSupplierConfiguration.java   | 30 ++++++++-----------
 2 files changed, 14 insertions(+), 17 deletions(-)

diff --git a/applications/source/rabbit-source/src/test/java/org/springframework/cloud/stream/app/source/rabbit/RabbitSourceListenerTests.java b/applications/source/rabbit-source/src/test/java/org/springframework/cloud/stream/app/source/rabbit/RabbitSourceListenerTests.java
index 02256f91..d68f06d1 100644
--- a/applications/source/rabbit-source/src/test/java/org/springframework/cloud/stream/app/source/rabbit/RabbitSourceListenerTests.java
+++ b/applications/source/rabbit-source/src/test/java/org/springframework/cloud/stream/app/source/rabbit/RabbitSourceListenerTests.java
@@ -77,6 +77,7 @@ public class RabbitSourceListenerTests {
 						"--spring.rabbitmq.listener.simple.acknowledgeMode=AUTO",
 						"--spring.rabbitmq.listener.simple.prefetch=10",
 						"--spring.rabbitmq.listener.simple.transactionSize=5",
+						"--spring.cloud.stream.function.bindings.rabbitSupplier-out-0=output",
 						"--spring.rabbitmq.port=" +
 								"${spring.rabbitmq.test.port}"
 				)) {
diff --git a/functions/supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierConfiguration.java b/functions/supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierConfiguration.java
index 919a3db3..ea5d7383 100644
--- a/functions/supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierConfiguration.java
+++ b/functions/supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierConfiguration.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2020 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -20,6 +20,7 @@ import java.util.function.Supplier;
 
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Envelope;
+import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 
 import org.springframework.amqp.core.AcknowledgeMode;
@@ -41,7 +42,7 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties
 import org.springframework.context.annotation.Bean;
 import org.springframework.integration.amqp.dsl.Amqp;
 import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
-import org.springframework.integration.channel.FluxMessageChannel;
+import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.messaging.Message;
 import org.springframework.retry.interceptor.RetryOperationsInterceptor;
 import org.springframework.util.Assert;
@@ -127,24 +128,19 @@ public class RabbitSupplierConfiguration implements DisposableBean {
 	}
 
 	@Bean
-	public AmqpInboundChannelAdapter adapter(SimpleMessageListenerContainer container,
-											FluxMessageChannel channel) {
-		return Amqp.inboundAdapter(container)
-				.autoStartup(false)
-				.outputChannel(channel)
-				.mappedRequestHeaders(properties.getMappedRequestHeaders())
-				.get();
+	public Publisher<Message<byte[]>> rabbitPublisher(SimpleMessageListenerContainer container) {
+		return IntegrationFlows.from(
+				Amqp.inboundAdapter(container)
+						.autoStartup(false)
+						.mappedRequestHeaders(properties.getMappedRequestHeaders()))
+				.toReactivePublisher();
 	}
 
 	@Bean
-	public Supplier<Flux<Message<?>>> rabbitSupplier(AmqpInboundChannelAdapter adapter,
-													FluxMessageChannel channel) {
-		return () -> Flux.from(channel).doOnSubscribe(subscription -> adapter.start());
-	}
-
-	@Bean
-	public FluxMessageChannel output() {
-		return new FluxMessageChannel();
+	public Supplier<Flux<Message<byte[]>>> rabbitSupplier(Publisher<Message<byte[]>> rabbitPublisher, AmqpInboundChannelAdapter adapter) {
+		return () -> Flux.from(rabbitPublisher)
+				.doOnSubscribe((subscription) -> adapter.start())
+				.doOnTerminate(adapter::stop);
 	}
 
 	@Bean