Refactor service bus queue multi binder sample (Azure#18946)

Author: wujack778

sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/README.md

@@ -2,11 +2,10 @@
 
 ## Key concepts
 This code sample demonstrates how to use the Spring Cloud Stream Binder for 
-multiple Azure Service Bus namespaces.
-In this sample you will bind to two Service Bus namespaces separately through 
-two queue binders.
-The sample app exposes RESTful APIs to receive string message.
-Then message is sent through Azure Service Bus to a `sink` which simply logs the message.
+multiple Azure Service Bus namespaces. In this sample you will bind to two Service Bus namespaces separately through 
+two queue binders.The sample app has two operating modes. One way is to expose a Restful API to receive string message,
+another way is to automatically provide string messages. These messages are published to a service bus.
+The sample will also consume messages from the same service bus.
 
 ## Getting started
 
@@ -42,14 +41,14 @@ is completed before the run.
       cloud:
         stream:
           bindings:
-            input:
+            consume1-in-0:
               destination: [servicebus-queue-1-name]
-            output:
+            supply1-out-0:
               destination: [servicebus-queue-1-name-same-as-above]
-            input1:
+            consume2-in-0:
               binder: servicebus-2
               destination: [servicebus-queue-2-name]
-            output1:
+            supply2-out-0:
               binder: servicebus-2
               destination: [servicebus-queue-2-name-same-as-above]
 
@@ -71,7 +70,15 @@ is completed before the run.
                   cloud:
                     azure:
                       servicebus:
-                        connection-string: [servicebus-namespace-2-connection-string]         
+                        connection-string: [servicebus-namespace-2-connection-string]
+    
+          #To specify which functional bean to bind to the external destination(s) exposed by the bindings
+          function:
+            definition: consume1;supply1;consume2;supply2;
+          poller:
+            initial-delay: 0
+            fixed-delay: 1000
+
     ```
 
 > The **defaultCandidate** configuration item:
@@ -80,26 +87,30 @@ default binder, or can be used only when explicitly referenced. This
 allows adding binder configurations without interfering with the default
 processing.
 
+>[!Important]
+>
+>  When using the Restful API to send messages, the **Active profiles** must contain `manual`.
+
 1.  Run the `mvn clean spring-boot:run` in the root of the code sample
     to get the app running.
 
 1.  Send a POST request to test the default binder
 
-        $ curl -X POST http://localhost:8080/messages?message=hello
+        $ curl -X POST http://localhost:8080/messages1?message=hello
 
 1.  Verify in your app’s logs that a similar message was posted:
 
-        [1] New message received: 'hello'
-        [1] Message 'hello' successfully checkpointed
+        [1] New message1 received: 'hello'
+        [1] Message1 'hello' successfully checkpointed
 
 1.  Send another POST request to test the other binder
 
-        $ curl -X POST http://localhost:8080/messages1?message=hello
+        $ curl -X POST http://localhost:8080/messages2?message=hello
 
 1.  Verify in your app’s logs that a similar message was posted:
 
-        [2] New message received: 'hello'
-        [2] Message 'hello' successfully checkpointed
+        [2] New message2 received: 'hello'
+        [2] Message2 'hello' successfully checkpointed
 
 6.  Delete the resources on [Azure Portal][azure-portal]
     to avoid unexpected charges.

sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/src/main/java/com/azure/spring/sample/servicebus/queue/multibinders/CustomProcessor.java

@@ -0,0 +1,46 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.spring.sample.servicebus.queue.multibinders;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Profile;
+import org.springframework.messaging.Message;
+import reactor.core.publisher.EmitterProcessor;
+import reactor.core.publisher.Flux;
+
+import java.util.function.Supplier;
+
+@Configuration
+@Profile("manual")
+public class ManualServiceProducerConfiguration {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(ServiceBusQueueMultiBindersApplication.class);
+
+    @Bean
+    public EmitterProcessor<Message<String>> emitterProcessor1() {
+        return EmitterProcessor.create();
+    }
+
+    @Bean
+    public EmitterProcessor<Message<String>> emitterProcessor2() {
+        return EmitterProcessor.create();
+    }
+
+    @Bean
+    public Supplier<Flux<Message<String>>> supply1(EmitterProcessor<Message<String>> emitterProcessor1) {
+        return () -> Flux.from(emitterProcessor1)
+                         .doOnNext(m -> LOGGER.info("Manually sending message1 {}", m))
+                         .doOnError(t -> LOGGER.error("Error encountered", t));
+    }
+
+    @Bean
+    public Supplier<Flux<Message<String>>> supply2(EmitterProcessor<Message<String>> emitterProcessor2) {
+        return () -> Flux.from(emitterProcessor2)
+                         .doOnNext(m -> LOGGER.info("Manually sending message2 {}", m))
+                         .doOnError(t -> LOGGER.error("Error encountered", t));
+    }
+}

sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/src/main/java/com/azure/spring/sample/servicebus/queue/multibinders/ManualServiceProducerConfiguration.java

@@ -3,17 +3,56 @@
 
 package com.azure.spring.sample.servicebus.queue.multibinders;
 
+import com.azure.spring.integration.core.api.Checkpointer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
+import org.springframework.messaging.Message;
+
+import java.util.function.Consumer;
+
+import static com.azure.spring.integration.core.AzureHeaders.CHECKPOINTER;
 
 /**
  * @author Yi Liu, 2020-4-30.
  */
 @SpringBootApplication
 public class ServiceBusQueueMultiBindersApplication {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(ServiceBusQueueMultiBindersApplication.class);
+
     public static void main(String[] args) {
         SpringApplication.run(ServiceBusQueueMultiBindersApplication.class, args);
     }
 
+    @Bean
+    public Consumer<Message<String>> consume1() {
+        return message -> {
+            Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
+            LOGGER.info("New message1 received: '{}'", message);
+            checkpointer.success().handle((r, ex) -> {
+                if (ex == null) {
+                    LOGGER.info("Message1 '{}' successfully checkpointed", message);
+                }
+                return null;
+            });
+        };
+    }
+
+    @Bean
+    public Consumer<Message<String>> consume2() {
+        return message -> {
+            Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
+            LOGGER.info("New message2 received: '{}'", message);
+            checkpointer.success().handle((r, ex) -> {
+                if (ex == null) {
+                    LOGGER.info("Message2 '{}' successfully checkpointed", message);
+                }
+                return null;
+            });
+        };
+    }
+
 }

sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/src/main/java/com/azure/spring/sample/servicebus/queue/multibinders/ServiceBusQueueMultiBindersApplication.java

@@ -0,0 +1,40 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.spring.sample.servicebus.queue.multibinders;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Profile;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+
+import java.util.function.Supplier;
+
+@Configuration
+@Profile("!manual")
+public class ServiceProducerConfiguration {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(ServiceBusQueueMultiBindersApplication.class);
+
+    private int i = 0;
+    private int j = 0;
+
+    @Bean
+    public Supplier<Message<String>> supply1() {
+        return () -> {
+            LOGGER.info("Sending message1, sequence1 " + i);
+            return MessageBuilder.withPayload("Hello world1, " + i++).build();
+        };
+    }
+
+    @Bean
+    public Supplier<Message<String>> supply2() {
+        return () -> {
+            LOGGER.info("Sending message2, sequence2 " + j);
+            return MessageBuilder.withPayload("Hello world2, " + j++).build();
+        };
+    }
+}

sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/src/main/java/com/azure/spring/sample/servicebus/queue/multibinders/ServiceProducerConfiguration.java

@@ -0,0 +1,44 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.spring.sample.servicebus.queue.multibinders;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Profile;
+import org.springframework.http.ResponseEntity;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+import reactor.core.publisher.EmitterProcessor;
+
+import javax.annotation.Resource;
+
+@RestController
+@Profile("manual")
+public class ServiceProducerController {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(ServiceBusQueueMultiBindersApplication.class);
+
+    @Resource(name = "emitterProcessor1")
+    private EmitterProcessor<Message<String>> emitterProcessor1;
+
+    @Resource(name = "emitterProcessor2")
+    private EmitterProcessor<Message<String>> emitterProcessor2;
+
+    @PostMapping("/messages1")
+    public ResponseEntity<String> sendMessage1(@RequestParam String message) {
+        LOGGER.info("Going to add message {} to emitter1", message);
+        emitterProcessor1.onNext(MessageBuilder.withPayload(message).build());
+        return ResponseEntity.ok("Sent1!");
+    }
+
+    @PostMapping("/messages2")
+    public ResponseEntity<String> sendMessage2(@RequestParam String message) {
+        LOGGER.info("Going to add message {} to emitter2", message);
+        emitterProcessor2.onNext(MessageBuilder.withPayload(message).build());
+        return ResponseEntity.ok("Sent2!");
+    }
+}