Refactoring multi binder samples

multi-binder-samples/multi-binder-kafka-rabbit/README.adoc

@@ -1,10 +1,10 @@
-== Spring Cloud Stream Multibinder Application with Different Systems
+== Spring Cloud Stream Multi-binder Application with Different Systems (Kafka on the inbound and RabbitMQ on the outbound)
 
 This example shows how to run a Spring Cloud Stream application with two different binder types (Kafka and RabbitMQ)
 
 ## Running the application
 
-The following instructions assume that you are running Kafka and RabbitMQ as a Docker images.
+The following instructions assume that you are running Kafka and RabbitMQ as Docker images.
 
 Go to the application root:
 

multi-binder-samples/multi-binder-kafka-rabbit/pom.xml

@@ -0,0 +1,126 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+
+	<artifactId>multi-binder-kafka-rabbit</artifactId>
+	<version>0.0.1-SNAPSHOT</version>
+	<packaging>jar</packaging>
+	<name>multi-binder-kafka-rabbit</name>
+	<description>Spring Cloud Stream Multibinder Sample</description>
+
+	<parent>
+		<groupId>org.springframework.boot</groupId>
+		<artifactId>spring-boot-starter-parent</artifactId>
+		<version>2.2.0.RELEASE</version>
+		<relativePath/> <!-- lookup parent from repository -->
+	</parent>
+
+	<properties>
+		<spring-cloud.version>Hoxton.BUILD-SNAPSHOT</spring-cloud.version>
+	</properties>
+
+	<dependencyManagement>
+		<dependencies>
+			<dependency>
+				<groupId>org.springframework.cloud</groupId>
+				<artifactId>spring-cloud-dependencies</artifactId>
+				<version>${spring-cloud.version}</version>
+				<type>pom</type>
+				<scope>import</scope>
+			</dependency>
+		</dependencies>
+	</dependencyManagement>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-binder-kafka</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-binder-rabbit</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-actuator</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-web</artifactId>
+		</dependency>
+	</dependencies>
+
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.springframework.boot</groupId>
+				<artifactId>spring-boot-maven-plugin</artifactId>
+			</plugin>
+		</plugins>
+	</build>
+
+	<repositories>
+		<repository>
+			<id>spring-snapshots</id>
+			<name>Spring Snapshots</name>
+			<url>https://repo.spring.io/libs-snapshot-local</url>
+			<snapshots>
+				<enabled>true</enabled>
+			</snapshots>
+			<releases>
+				<enabled>false</enabled>
+			</releases>
+		</repository>
+		<repository>
+			<id>spring-milestones</id>
+			<name>Spring Milestones</name>
+			<url>https://repo.spring.io/libs-milestone-local</url>
+			<snapshots>
+				<enabled>false</enabled>
+			</snapshots>
+		</repository>
+	</repositories>
+	<pluginRepositories>
+		<pluginRepository>
+			<id>spring-snapshots</id>
+			<name>Spring Snapshots</name>
+			<url>https://repo.spring.io/libs-snapshot-local</url>
+			<snapshots>
+				<enabled>true</enabled>
+			</snapshots>
+			<releases>
+				<enabled>false</enabled>
+			</releases>
+		</pluginRepository>
+		<pluginRepository>
+			<id>spring-milestones</id>
+			<name>Spring Milestones</name>
+			<url>https://repo.spring.io/libs-milestone-local</url>
+			<snapshots>
+				<enabled>false</enabled>
+			</snapshots>
+		</pluginRepository>
+		<pluginRepository>
+			<id>spring-releases</id>
+			<name>Spring Releases</name>
+			<url>https://repo.spring.io/libs-release-local</url>
+			<snapshots>
+				<enabled>false</enabled>
+			</snapshots>
+		</pluginRepository>
+	</pluginRepositories>
+</project>

multi-binder-samples/multi-binder-kafka-rabbit/src/main/java/multibinder/MultibinderApplication.java

@@ -16,8 +16,16 @@
 
 package multibinder;
 
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
 
 @SpringBootApplication
 public class MultibinderApplication {
@@ -26,4 +34,28 @@ public class MultibinderApplication {
 		SpringApplication.run(MultibinderApplication.class, args);
 	}
 
+	@Bean
+	public Function<String, String> process() {
+		return payload -> payload.toUpperCase();
+	}
+
+	static class TestProducer {
+
+		private AtomicBoolean semaphore = new AtomicBoolean(true);
+
+		@Bean
+		public Supplier<String> sendTestData() {
+			return () -> this.semaphore.getAndSet(!this.semaphore.get()) ? "foo" : "bar";
+		}
+	}
+
+	static class TestConsumer {
+
+		private final Log logger = LogFactory.getLog(getClass());
+
+		@Bean
+		public Consumer<String> receive() {
+			return s -> logger.info("Data received..." + s);
+		}
+	}
 }

multi-binder-samples/multi-binder-kafka-rabbit/src/main/resources/application.yml

@@ -2,18 +2,20 @@ spring:
   cloud:
     stream:
       bindings:
-        input:
+        process-in-0:
           destination: dataIn
           binder: kafka
-        output:
+        process-out-0:
           destination: dataOut
           binder: rabbit
         #Test sink binding (used for testing)
-        output1:
+        sendTestData-out-0:
           destination: dataIn
           binder: kafka
         #Test sink binding (used for testing)
-        input1:
+        receive-in-0:
           destination: dataOut
           binder: rabbit
+      function:
+        definition: sendTestData;process;receive
 

multi-binder-samples/pom.xml

@@ -2,14 +2,14 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
     <groupId>io.spring.cloud.stream.sample</groupId>
-    <artifactId>multibinder-samples</artifactId>
+    <artifactId>multi-binder-samples</artifactId>
     <version>0.0.1-SNAPSHOT</version>
     <packaging>pom</packaging>
     <name>multibinder-samples</name>
     <description>Collection of Spring Cloud Stream Aggregate Samples</description>
 
     <modules>
-        <module>multibinder-kafka-rabbit</module>
+        <module>multi-binder-kafka-rabbit</module>
         <module>multibinder-two-kafka-clusters</module>
         <module>kafka-multibinder-jaas</module>
         <module>multibinder-kafka-streams</module>

multibinder-samples/multibinder-kafka-rabbit/pom.xml

@@ -1,51 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
-	<modelVersion>4.0.0</modelVersion>
-
-	<artifactId>multibinder-kafka-rabbit</artifactId>
-	<version>0.0.1-SNAPSHOT</version>
-	<packaging>jar</packaging>
-	<name>multibinder-kafka-rabbit</name>
-	<description>Spring Cloud Stream Multibinder Sample</description>
-
-	<parent>
-		<groupId>io.spring.cloud.stream.sample</groupId>
-		<artifactId>spring-cloud-stream-samples-parent</artifactId>
-		<version>0.0.1-SNAPSHOT</version>
-		<relativePath>../..</relativePath>
-	</parent>
-
-	<dependencies>
-		<dependency>
-			<groupId>org.springframework.cloud</groupId>
-			<artifactId>spring-cloud-stream-binder-rabbit</artifactId>
-		</dependency>
-		<dependency>
-			<groupId>org.springframework.cloud</groupId>
-			<artifactId>spring-cloud-stream-binder-kafka</artifactId>
-		</dependency>
-		<dependency>
-			<groupId>org.springframework.cloud</groupId>
-			<artifactId>spring-cloud-stream-binder-rabbit-test-support</artifactId>
-			<scope>test</scope>
-		</dependency>
-		<dependency>
-			<groupId>org.springframework.boot</groupId>
-			<artifactId>spring-boot-starter-test</artifactId>
-			<scope>test</scope>
-		</dependency>
-        <dependency>
-            <groupId>org.springframework.kafka</groupId>
-            <artifactId>spring-kafka-test</artifactId>
-        </dependency>
-	</dependencies>
-
-	<build>
-		<plugins>
-			<plugin>
-				<groupId>org.springframework.boot</groupId>
-				<artifactId>spring-boot-maven-plugin</artifactId>
-			</plugin>
-		</plugins>
-	</build>
-</project>

multibinder-samples/multibinder-kafka-rabbit/src/test/java/multibinder/RabbitAndKafkaBinderApplicationTests.java

@@ -1,108 +0,0 @@
-/*
- * Copyright 2015-2017 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
- *
- *      https://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 multibinder;
-
-import java.util.UUID;
-
-import org.hamcrest.CoreMatchers;
-import org.hamcrest.Matchers;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.ClassRule;
-import org.junit.Ignore;
-import org.junit.Test;
-
-import org.springframework.amqp.rabbit.core.RabbitAdmin;
-import org.springframework.boot.SpringApplication;
-import org.springframework.cloud.stream.binder.BinderFactory;
-import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
-import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
-import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder;
-import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
-import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder;
-import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties;
-import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport;
-import org.springframework.context.ConfigurableApplicationContext;
-import org.springframework.integration.channel.DirectChannel;
-import org.springframework.integration.channel.QueueChannel;
-import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
-import org.springframework.messaging.Message;
-import org.springframework.messaging.MessageChannel;
-import org.springframework.messaging.support.MessageBuilder;
-import org.springframework.test.annotation.DirtiesContext;
-
-/**
- * @author Marius Bogoevici
- * @author Gary Russell
- */
-@DirtiesContext
-@Ignore
-public class RabbitAndKafkaBinderApplicationTests {
-
-	@ClassRule
-	public static RabbitTestSupport rabbitTestSupport = new RabbitTestSupport();
-
-	@ClassRule
-	public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, "test");
-
-
-	private final String randomGroup = UUID.randomUUID().toString();
-
-	@After
-	public void cleanUp() {
-		RabbitAdmin admin = new RabbitAdmin(rabbitTestSupport.getResource());
-		admin.deleteQueue("binder.dataOut.default");
-		admin.deleteQueue("binder.dataOut." + this.randomGroup);
-		admin.deleteExchange("binder.dataOut");
-	}
-
-	@Test
-	public void contextLoads() throws Exception {
-		// passing connection arguments arguments to the embedded Kafka instance
-		ConfigurableApplicationContext context = SpringApplication.run(MultibinderApplication.class,
-				"--spring.cloud.stream.kafka.binder.brokers=" + embeddedKafka.getEmbeddedKafka().getBrokersAsString());
-		context.close();
-	}
-
-	@Test
-	public void messagingWorks() throws Exception {
-		// passing connection arguments arguments to the embedded Kafka instance
-		ConfigurableApplicationContext context = SpringApplication.run(MultibinderApplication.class,
-				"--spring.cloud.stream.kafka.binder.brokers=" + embeddedKafka.getEmbeddedKafka().getBrokersAsString(),
-				"--spring.cloud.stream.bindings.input.group=testGroup",
-				"--spring.cloud.stream.bindings.output.producer.requiredGroups=" + this.randomGroup);
-		DirectChannel dataProducer = new DirectChannel();
-		BinderFactory binderFactory = context.getBean(BinderFactory.class);
-
-		QueueChannel dataConsumer = new QueueChannel();
-
-		((RabbitMessageChannelBinder) binderFactory.getBinder("rabbit", MessageChannel.class)).bindConsumer("dataOut", this.randomGroup,
-				dataConsumer, new ExtendedConsumerProperties<>(new RabbitConsumerProperties()));
-
-		((KafkaMessageChannelBinder) binderFactory.getBinder("kafka", MessageChannel.class))
-				.bindProducer("dataIn", dataProducer, new ExtendedProducerProperties<>(new KafkaProducerProperties()));
-
-		String testPayload = "testFoo" + this.randomGroup;
-		dataProducer.send(MessageBuilder.withPayload(testPayload).build());
-
-		Message<?> receive = dataConsumer.receive(60_000);
-		Assert.assertThat(receive, Matchers.notNullValue());
-		Assert.assertThat(receive.getPayload(), CoreMatchers.equalTo(testPayload));
-		context.close();
-	}
-
-}

multibinder-samples/multibinder-two-kafka-clusters/src/main/java/multibinder/BridgeTransformer.java

@@ -1,94 +0,0 @@
-/*
- * Copyright 2015 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
- *
- *      https://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 multibinder;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.springframework.cloud.stream.annotation.EnableBinding;
-import org.springframework.cloud.stream.annotation.Input;
-import org.springframework.cloud.stream.annotation.Output;
-import org.springframework.cloud.stream.annotation.StreamListener;
-import org.springframework.cloud.stream.messaging.Processor;
-import org.springframework.context.annotation.Bean;
-import org.springframework.integration.annotation.InboundChannelAdapter;
-import org.springframework.integration.annotation.Poller;
-import org.springframework.integration.core.MessageSource;
-import org.springframework.messaging.MessageChannel;
-import org.springframework.messaging.SubscribableChannel;
-import org.springframework.messaging.handler.annotation.SendTo;
-import org.springframework.messaging.support.GenericMessage;
-
-import java.util.concurrent.atomic.AtomicBoolean;
-
-/**
- * @author Marius Bogoevici
- * @author Soby Chacko
- */
-@EnableBinding(Processor.class)
-public class BridgeTransformer {
-
-	@StreamListener(Processor.INPUT)
-	@SendTo(Processor.OUTPUT)
-	public Object transform(Object payload) {
-		return payload;
-	}
-
-	//Following source is used as test producer.
-	@EnableBinding(TestSource.class)
-	static class TestProducer {
-
-		private AtomicBoolean semaphore = new AtomicBoolean(true);
-
-		@Bean
-		@InboundChannelAdapter(channel = TestSource.OUTPUT, poller = @Poller(fixedDelay = "1000"))
-		public MessageSource<String> sendTestData() {
-			return () ->
-					new GenericMessage<>(this.semaphore.getAndSet(!this.semaphore.get()) ? "foo" : "bar");
-
-		}
-	}
-
-	//Following sink is used as test consumer for the above processor. It logs the data received through the processor.
-	@EnableBinding(TestSink.class)
-	static class TestConsumer {
-
-		private final Log logger = LogFactory.getLog(getClass());
-
-		@StreamListener(TestSink.INPUT)
-		public void receive(String data) {
-			logger.info("Data received..." + data);
-		}
-	}
-
-	interface TestSink {
-
-		String INPUT = "input1";
-
-		@Input(INPUT)
-		SubscribableChannel input1();
-
-	}
-
-	interface TestSource {
-
-		String OUTPUT = "output1";
-
-		@Output(TestSource.OUTPUT)
-		MessageChannel output();
-
-	}
-}

pom.xml

@@ -24,8 +24,8 @@
 		<module>processor-samples</module>
 		<module>kafka-streams-samples</module>
 		<module>multi-functions-samples</module>
+		<module>multi-binder-samples</module>
 		<module>kinesis-samples</module>
-		<module>multibinder-samples</module>
 		<module>schema-registry-samples</module>
 		<module>partitioning-samples</module>
 		<module>transaction-kafka-samples</module>