Refactor service bus queue binder sample with functional approach (Azure#18920)

Author: wujack778

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

@@ -3,8 +3,9 @@
 ## Key concepts
 
 This code sample demonstrates how to use the Spring Cloud Stream binder for 
-Azure Service Bus queue. The sample app exposes a RESTful API to receive string message.
-Then message is sent through Azure Service Bus to a `sink` which simply logs the message.
+Azure Service Bus queue. 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 queue. The sample will also consume messages from the same service bus queue.
 
 ## Getting started
 
@@ -18,9 +19,12 @@ completed before the run.
 
 ### Create Azure resources
 
-We have several ways to config the Spring Cloud Stream Binder for Azure
-Event Hub. You can choose anyone of them.
+We have several ways to config the Spring Cloud Stream Binder for Service
+Bus Queue. You can choose anyone of them.
 
+>[!Important]
+>
+>  When using the Restful API to send messages, the **Active profiles** must contain `manual`.
 
 #### Method 1: Connection string based usage
 
@@ -36,10 +40,15 @@ Event Hub. You can choose anyone of them.
             connection-string: [servicebus-namespace-connection-string] 
         stream:
           bindings: 
-            input: 
+            consume-in-0: 
               destination: [servicebus-queue-name]
-            output:
+            supply-out-0:
               destination: [servicebus-queue-name-same-as-above]
+          function:
+            definition: consume;supply;
+          poller:
+            fixed-delay: 1000
+            initial-delay: 0
     ```
 
 #### Method 2: Service principal based usage
@@ -49,6 +58,10 @@ Event Hub. You can choose anyone of them.
 
 1.  Create Azure Service Bus namespace and queue.
         Please see [how to create][create-service-bus].
+
+1.  Add Role Assignment for Service Bus. See
+    [Service principal for Azure resources with Service Bus][role-assignment]
+    to add role assignment for Service Bus. Assign `Contributor` role for service bus.
 
 1.  Update [application-sp.yaml][application-sp.yaml].
     ```yaml
@@ -63,12 +76,16 @@ Event Hub. You can choose anyone of them.
             namespace: [servicebus-namespace]
         stream:
           bindings:
-            input:
+            consume-in-0:
               destination: [servicebus-queue-name]
-            output:
+            supply-out-0:
               destination: [servicebus-queue-name-same-as-above]
+          function:
+            definition: consume;supply;
+          poller:
+            fixed-delay: 1000
+            initial-delay: 0
     ```
-        
 #### Method 3: MSI credential based usage
 
 ##### Set up managed identity
@@ -100,10 +117,15 @@ Please follow [create managed identity][create-managed-identity] to set up manag
             namespace: [servicebus-namespace]
         stream:
           bindings:
-            input:
+            consume-in-0:
               destination: [servicebus-queue-name]
-            output:
+            supply-out-0:
               destination: [servicebus-queue-name-same-as-above]
+          function:
+            definition: consume;supply;
+          poller:
+            fixed-delay: 1000
+            initial-delay: 0
     ```
     > We should specify `spring.profiles.active=mi` to run the Spring Boot application. 
       For App Service, please add a configuration entry for this.

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

@@ -0,0 +1,34 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.spring.sample.servicebus.queue.binder;
+
+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(ServiceBusQueueBinderApplication.class);
+
+    @Bean
+    public EmitterProcessor<Message<String>> emitter() {
+        return EmitterProcessor.create();
+    }
+
+    @Bean
+    public Supplier<Flux<Message<String>>> supply(EmitterProcessor<Message<String>> emitter) {
+        return () -> Flux.from(emitter)
+                         .doOnNext(m -> LOGGER.info("Manually sending message {}", m))
+                         .doOnError(t -> LOGGER.error("Error encountered", t));
+    }
+}

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

@@ -3,16 +3,38 @@
 
 package com.azure.spring.sample.servicebus.queue.binder;
 
+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 Warren Zhu
- */
 @SpringBootApplication
 public class ServiceBusQueueBinderApplication {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(ServiceBusQueueBinderApplication.class);
+
     public static void main(String[] args) {
         SpringApplication.run(ServiceBusQueueBinderApplication.class, args);
     }
+
+    @Bean
+    public Consumer<Message<String>> consume() {
+        return message -> {
+            Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
+            LOGGER.info("New message received: '{}'", message);
+            checkpointer.success().handle((r, ex) -> {
+                if (ex == null) {
+                    LOGGER.info("Message '{}' successfully checkpointed", message);
+                }
+                return null;
+            });
+        };
+    }
 }

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

@@ -0,0 +1,31 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.spring.sample.servicebus.queue.binder;
+
+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(ServiceBusQueueBinderApplication.class);
+
+    private int i = 0;
+
+    @Bean
+    public Supplier<Message<String>> supply() {
+        return () -> {
+            LOGGER.info("Sending message, sequence " + i);
+            return MessageBuilder.withPayload("Hello world, " + i++).build();
+        };
+    }
+}

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

@@ -0,0 +1,39 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.spring.sample.servicebus.queue.binder;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+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.GetMapping;
+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;
+
+@RestController
+@Profile("manual")
+public class ServiceProducerController {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(ServiceBusQueueBinderApplication.class);
+
+    @Autowired
+    private EmitterProcessor<Message<String>> emitterProcessor;
+
+    @PostMapping("/messages")
+    public ResponseEntity<String> sendMessage(@RequestParam String message) {
+        LOGGER.info("Going to add message {} to emitter", message);
+        emitterProcessor.onNext(MessageBuilder.withPayload(message).build());
+        return ResponseEntity.ok("Sent!");
+    }
+
+    @GetMapping("/")
+    public String welcome() {
+        return "welcome";
+    }
+}

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

@@ -16,13 +16,18 @@ spring:
         namespace: [servicebus-namespace]
     stream:
       bindings:
-        input:
+        consume-in-0:
           destination: [servicebus-queue-name]
-        output:
+        supply-out-0:
           destination: [servicebus-queue-name-same-as-above]
       servicebus:
         queue:
           bindings:
-            input:
+            consume-in-0:
               consumer:
-                checkpoint-mode: MANUAL
+                checkpoint-mode: MANUAL
+      function:
+        definition: consume;supply;
+      poller:
+        fixed-delay: 1000
+        initial-delay: 0

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

@@ -17,13 +17,18 @@ spring:
         namespace: [servicebus-namespace]
     stream:
       bindings:
-        input:
+        consume-in-0:
           destination: [servicebus-queue-name]
-        output:
+        supply-out-0:
           destination: [servicebus-queue-name-same-as-above]
       servicebus:
         queue:
           bindings:
-            input:
+            consume-in-0:
               consumer:
-                checkpoint-mode: MANUAL
+                checkpoint-mode: MANUAL
+      function:
+        definition: consume;supply;
+      poller:
+        fixed-delay: 1000
+        initial-delay: 0

sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-binder/src/main/resources/application-mi.yaml

@@ -5,13 +5,18 @@ spring:
         connection-string: [servicebus-namespace-connection-string]
     stream:
       bindings:
-        input:
+        consume-in-0:
           destination: [servicebus-queue-name]
-        output:
+        supply-out-0:
           destination: [servicebus-queue-name-same-as-above]
       servicebus:
         queue:
           bindings:
-            input:
+            consume-in-0:
               consumer:
-                checkpoint-mode: MANUAL
+                checkpoint-mode: MANUAL
+      function:
+        definition: consume;supply;
+      poller:
+        fixed-delay: 1000
+        initial-delay: 0