Replace @Streamlistener with functional approach in eventhub-binder sample (Azure#18800)

Author: wujack778

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

@@ -2,10 +2,9 @@
 
 ## Key concepts
 
-This code sample demonstrates how to use the Spring Cloud Stream Binder
-for Azure Event Hub. The sample app exposes a RESTful API to receive string
-message. Then message is sent through Azure Event Hub to a `sink` which
-simply logs the message.
+This code sample demonstrates how to use the Spring Cloud Stream Binder for Azure Event Hub.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 an event hub. The sample will also consume messages from the same event hub.
 
 ## Getting started
 
@@ -22,6 +21,10 @@ completed before the run.
 We have several ways to config the Spring Cloud Stream Binder for Azure
 Event Hub. 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
 
@@ -46,10 +49,10 @@ Event Hub. You can choose anyone of them.
             checkpoint-container: [checkpoint-container]
         stream:
           bindings:
-            input:
+            consume-in-0:
               destination: [eventhub-name]
               group: [consumer-group]
-            output:
+            supply-out-0:
               destination: [the-same-eventhub-name-as-above]
     ```
 
@@ -63,6 +66,15 @@ Event Hub. You can choose anyone of them.
     can create your own Consumer Group or use the default "$Default" Consumer Group.
 
 1.  Create [Azure Storage][create-azure-storage] for checkpoint use.
+
+1.  Add Role Assignment for Event Hub, Storage Account and Resource group. See
+    [Service principal for Azure resources with Event Hubs][role-assignment]
+    to add role assignment for Event Hub, Storage Account, Resource group is similar.
+
+    - Resource group: assign `Contributor` role for service principal.
+    - Event Hub: assign `Contributor` role for service principal.
+    - Storage Account: assign `Storage Account Key Operator Service Role` 
+      role for service principal.
 
 1.  Update [application-sp.yaml][application-sp.yaml].
     ```yaml
@@ -79,12 +91,13 @@ Event Hub. You can choose anyone of them.
             checkpoint-container: [checkpoint-container]
         stream:
           bindings:
-            input:
+            consume-in-0:
               destination: [eventhub-name]
               group: [consumer-group]
-            output:
+            supply-out-0:
               destination: [the-same-eventhub-name-as-above]
     ```
+    > We should specify `spring.profiles.active=sp` to run the Spring Boot application. 
 
 #### Method 3: MSI credential based usage
 
@@ -127,10 +140,10 @@ Please follow [create managed identity][create-managed-identity] to set up manag
             checkpoint-container: [checkpoint-container]
         stream:
           bindings:
-            input:
+            consume-in-0:
               destination: [eventhub-name]
               group: [consumer-group]
-            output:
+            supply-out-0:
               destination: [the-same-eventhub-name-as-above]
     ```
     > We should specify `spring.profiles.active=mi` to run the Spring Boot application. 

sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/main/java/com/azure/spring/sample/eventhubs/binder/EventHubBinderApplication.java

@@ -3,16 +3,53 @@
 
 package com.azure.spring.sample.eventhubs.binder;
 
+import com.azure.spring.integration.core.api.reactor.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.integration.annotation.ServiceActivator;
+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 EventHubBinderApplication {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(EventHubBinderApplication.class);
+
     public static void main(String[] args) {
         SpringApplication.run(EventHubBinderApplication.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()
+                        .doOnSuccess(success -> LOGGER.info("Message '{}' successfully checkpointed", message))
+                        .doOnError(error -> LOGGER.error("Exception: {}", error.getMessage()))
+                        .subscribe();
+        };
+    }
+
+    // Replace destination with spring.cloud.stream.bindings.consume-in-0.destination
+    // Replace group with spring.cloud.stream.bindings.consume-in-0.group
+    @ServiceActivator(inputChannel = "{destination}.{group}.errors")
+    public void consumerError(Message<?> message) {
+        LOGGER.error("Handling customer ERROR: " + message);
+    }
+
+    // Replace destination with spring.cloud.stream.bindings.supply-out-0.destination
+    @ServiceActivator(inputChannel = "{destination}.errors")
+    public void producerError(Message<?> message) {
+        LOGGER.error("Handling Producer ERROR: " + message);
+    }
 }

sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/main/java/com/azure/spring/sample/eventhubs/binder/EventProducerConfiguration.java

@@ -0,0 +1,32 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.spring.sample.eventhubs.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 EventProducerConfiguration {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(EventHubBinderApplication.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-eventhubs-binder/src/main/java/com/azure/spring/sample/eventhubs/binder/EventProducerController.java

@@ -0,0 +1,43 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.spring.sample.eventhubs.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;
+
+/**
+ * @author Warren Zhu
+ */
+@RestController
+@Profile("manual")
+public class EventProducerController {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(EventHubBinderApplication.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-eventhubs-binder/src/main/java/com/azure/spring/sample/eventhubs/binder/ManualEventProducerConfiguration.java

@@ -0,0 +1,34 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.spring.sample.eventhubs.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 ManualEventProducerConfiguration {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(EventHubBinderApplication.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-eventhubs-binder/src/main/java/com/azure/spring/sample/eventhubs/binder/SinkExample.java

@@ -18,17 +18,23 @@ spring:
         checkpoint-container: [checkpoint-container]
     stream:
       bindings:
-        input:
+        consume-in-0:
           destination: [eventhub-name]
           group: [consumer-group]
-        output:
+        supply-out-0:
           destination: [the-same-eventhub-name-as-above]
 
       eventhub:
         bindings:
-          input:
+          consume-in-0:
             consumer:
               checkpoint-mode: MANUAL
       default:
         producer:
           errorChannelEnabled: true
+      function:
+        definition: consume;supply;
+      poller:
+        initial-delay: 0
+        fixed-delay: 1000
+

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

@@ -19,19 +19,23 @@ spring:
         checkpoint-container: [checkpoint-container]
     stream:
       bindings:
-        input:
+        consume-in-0:
           destination: [eventhub-name]
           group: [consumer-group]
-        output:
+        supply-out-0:
           destination: [the-same-eventhub-name-as-above]
 
       eventhub:
         bindings:
-          input:
+          consume-in-0:
             consumer:
               checkpoint-mode: MANUAL
       default:
         producer:
           errorChannelEnabled: true
-
+      function:
+        definition: consume;supply;
+      poller:
+        initial-delay: 0
+        fixed-delay: 1000