Refactor reactive processor sample application

processor-samples/reactive-processor-rabbit/src/main/java/reactive/kafka/ReactiveProcessorApplication.java

@@ -0,0 +1,62 @@
+package reactive.kafka;
+
+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.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.support.GenericMessage;
+import reactor.core.publisher.Flux;
+
+import java.time.Duration;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+@SpringBootApplication
+public class ReactiveProcessorApplication {
+
+	private final Log logger = LogFactory.getLog(getClass());
+
+	private AtomicBoolean semaphore = new AtomicBoolean(true);
+
+	public static void main(String[] args) {
+		SpringApplication.run(ReactiveProcessorApplication.class, args);
+	}
+
+	@Bean
+	public Function<Flux<String>, Flux<String>> aggregate() {
+		return inbound -> inbound.
+				log()
+				.window(Duration.ofSeconds(30), Duration.ofSeconds(5))
+				.flatMap(w -> w.reduce("", (s1,s2)->s1+s2))
+				.log();
+	}
+
+	//Following source and sinks are used for testing only.
+	//Test source will send data to the same destination where the processor receives data
+	//Test sink will consume data from the same destination where the processor produces data
+
+	@Bean
+	public Supplier<String> testSource() {
+		return () -> this.semaphore.getAndSet(!this.semaphore.get()) ? "foo" : "bar";
+
+	}
+
+	@Bean
+	public Consumer<String>  testSink() {
+		return payload -> logger.info("Data received: " + payload);
+
+	}
+}

processor-samples/reactive-processor-rabbit/src/main/resources/application.yml

@@ -0,0 +1,13 @@
+spring:
+  cloud:
+    stream:
+      function.definition: aggregate;testSource;testSink
+      bindings:
+        aggregate-out-0:
+          destination: transformed
+        testSink-in-0:
+          destination: transformed
+        aggregate-in-0:
+          destination: testtock
+        testSource-out-0:
+          destination: testtock