Update spring cloud sample for bug modification and api upgrade

sdk/spring/azure-spring-boot-samples/azure-spring-boot-sample-servicebus/src/main/java/com/azure/spring/sample/servicebus/ServiceBusSampleApplication.java

@@ -7,6 +7,8 @@
 import com.azure.messaging.servicebus.ServiceBusMessage;
 import com.azure.messaging.servicebus.ServiceBusReceiverAsyncClient;
 import com.azure.messaging.servicebus.ServiceBusSenderAsyncClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.CommandLineRunner;
 import org.springframework.boot.SpringApplication;
@@ -18,6 +20,7 @@
 
 @SpringBootApplication
 public class ServiceBusSampleApplication implements CommandLineRunner {
+    private static final Logger LOGGER = LoggerFactory.getLogger(ServiceBusSampleApplication.class);
 
     @Autowired
     private ServiceBusSenderAsyncClient queueSender;
@@ -49,9 +52,9 @@ private void sendQueueMessage() throws InterruptedException {
         final String messageBody = "queue message";
 
         queueSender.sendMessage(new ServiceBusMessage(BinaryData.fromBytes(messageBody.getBytes(UTF_8)))).subscribe(
-            v -> System.out.println("Sent message: " + messageBody),
-            e -> System.err.println("Error occurred while sending message: " + e),
-            () -> System.out.println("Send message to queue complete.")
+            v -> LOGGER.info("Sent message: {}", messageBody),
+            e -> LOGGER.error("Error occurred while sending message", e),
+            () -> LOGGER.info("Send message to queue complete.")
         );
 
         TimeUnit.SECONDS.sleep(5);
@@ -61,7 +64,7 @@ private void sendQueueMessage() throws InterruptedException {
 
     private void receiveQueueMessage() throws InterruptedException {
         queueReceiver.receiveMessages().subscribe(message ->
-            System.out.println("Received Message: " + message.getBody().toString()));
+            LOGGER.info("Received Message: {}", message.getBody().toString()));
 
         TimeUnit.SECONDS.sleep(5);
 
@@ -72,9 +75,9 @@ private void sendTopicMessage() throws InterruptedException {
         final String messageBody = "topic message";
 
         topicSender.sendMessage(new ServiceBusMessage(BinaryData.fromBytes(messageBody.getBytes(UTF_8)))).subscribe(
-            v -> System.out.println("Sent message: " + messageBody),
-            e -> System.err.println("Error occurred while sending message: " + e),
-            () -> System.out.println("Send message to topic complete.")
+            v -> LOGGER.info("Sent message: {}", messageBody),
+            e -> LOGGER.error("Error occurred while sending message", e),
+            () -> LOGGER.info("Send message to topic complete.")
         );
 
         TimeUnit.SECONDS.sleep(10);
@@ -84,7 +87,7 @@ private void sendTopicMessage() throws InterruptedException {
 
     private void receiveSubscriptionMessage() throws InterruptedException {
         topicSubscriber.receiveMessages().subscribe(message ->
-            System.out.println("Received Message: " + message.getBody().toString()));
+            LOGGER.info("Received Message: {}", message.getBody().toString()));
 
         TimeUnit.SECONDS.sleep(10);
 

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

@@ -46,6 +46,8 @@ Event Hub. You can choose anyone of them.
             checkpoint-access-key: [checkpoint-access-key]
             checkpoint-container: [checkpoint-container]
         stream:
+          function:
+            definition: consume;supply
           bindings:
             consume-in-0:
               destination: [eventhub-name]
@@ -88,6 +90,8 @@ Event Hub. You can choose anyone of them.
             checkpoint-storage-account: [checkpoint-storage-account]
             checkpoint-container: [checkpoint-container]
         stream:
+          function:
+            definition: consume;supply
           bindings:
             consume-in-0:
               destination: [eventhub-name]
@@ -137,6 +141,8 @@ Please follow [create managed identity][create-managed-identity] to set up manag
             checkpoint-storage-account: [checkpoint-storage-account]
             checkpoint-container: [checkpoint-container]
         stream:
+          function:
+            definition: consume;supply
           bindings:
             consume-in-0:
               destination: [eventhub-name]

sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/pom.xml

@@ -21,7 +21,12 @@
         <version>2020.0.1</version> <!-- {x-version-update;org.springframework.cloud:spring-cloud-dependencies;external_dependency} -->
         <type>pom</type>  
         <scope>import</scope> 
-      </dependency> 
+      </dependency>
+      <dependency>
+        <groupId>com.azure</groupId>
+        <artifactId>azure-core</artifactId>
+        <version>1.14.1</version> <!-- {x-version-update;com.azure:azure-core;dependency} -->
+      </dependency>
     </dependencies> 
   </dependencyManagement>  
   <dependencies> 

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

@@ -43,7 +43,7 @@ public Consumer<Message<String>> consume() {
 
             checkpointer.success()
                         .doOnSuccess(success -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
-                        .doOnError(error -> LOGGER.error("Exception: {}", error.getMessage()))
+                        .doOnError(error -> LOGGER.error("Exception found", error))
                         .subscribe();
         };
     }

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

@@ -14,7 +14,7 @@
 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 reactor.core.publisher.Sinks;
 
 /**
  * @author Warren Zhu
@@ -26,13 +26,13 @@ public class EventProducerController {
     private static final Logger LOGGER = LoggerFactory.getLogger(EventHubBinderApplication.class);
 
     @Autowired
-    private EmitterProcessor<Message<String>> emitterProcessor;
+    private Sinks.Many<Message<String>> many;
 
     @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!");
+        LOGGER.info("Going to add message {} to Sinks.Many.", message);
+        many.emitNext(MessageBuilder.withPayload(message).build(), Sinks.EmitFailureHandler.FAIL_FAST);
+        return ResponseEntity.ok(message);
     }
 
     @GetMapping("/")

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

@@ -9,8 +9,8 @@
 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 reactor.core.publisher.Sinks;
 
 import java.util.function.Supplier;
 
@@ -21,13 +21,13 @@ public class ManualEventProducerConfiguration {
     private static final Logger LOGGER = LoggerFactory.getLogger(EventHubBinderApplication.class);
 
     @Bean
-    public EmitterProcessor<Message<String>> emitter() {
-        return EmitterProcessor.create();
+    public Sinks.Many<Message<String>> many() {
+        return Sinks.many().unicast().onBackpressureBuffer();
     }
 
     @Bean
-    public Supplier<Flux<Message<String>>> supply(EmitterProcessor<Message<String>> emitter) {
-        return () -> Flux.from(emitter)
+    public Supplier<Flux<Message<String>>> supply(Sinks.Many<Message<String>> many) {
+        return () -> many.asFlux()
                          .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/resources/application-mi.yaml

@@ -17,6 +17,8 @@ spring:
         checkpoint-storage-account: [checkpoint-storage-account]
         checkpoint-container: [checkpoint-container]
     stream:
+      function:
+        definition: consume;supply
       bindings:
         consume-in-0:
           destination: [eventhub-name]
@@ -32,8 +34,6 @@ spring:
       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/resources/application-sp.yaml

@@ -18,6 +18,8 @@ spring:
         checkpoint-storage-account: [checkpoint-storage-account]
         checkpoint-container: [checkpoint-container]
     stream:
+      function:
+        definition: consume;supply
       bindings:
         consume-in-0:
           destination: [eventhub-name]
@@ -33,8 +35,6 @@ spring:
       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/resources/application.yaml

@@ -8,6 +8,8 @@ spring:
         checkpoint-container: [checkpoint-container]
 
     stream:
+      function:
+        definition: consume;supply
       bindings:
         consume-in-0:
           destination: [eventhub-name]
@@ -23,8 +25,6 @@ spring:
       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-kafka/README.md

@@ -4,7 +4,7 @@
 
 This code sample demonstrates how to use the Spring Cloud Stream Kafka
 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`
+string message. Then message is sent through Azure Event Hub to a bean `consumer`
 which simply logs the message.
 
 ## Getting started
@@ -42,11 +42,13 @@ Running this sample will be charged by Azure. You can check the usage and bill a
           eventhub:
             namespace: [eventhub-namespace]
         stream:
+          function:
+            definition: consume;supply
           bindings:
-            input:
+            consume-in-0:
               destination: [eventhub-name]
               group: [consumer-group]
-            output:
+            supply-out-0:
               destination: [the-same-eventhub-name-as-above]
     ```
 

sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/pom.xml

@@ -23,6 +23,11 @@
         <type>pom</type>
         <scope>import</scope>
       </dependency>
+      <dependency>
+        <groupId>com.azure</groupId>
+        <artifactId>azure-core</artifactId>
+        <version>1.14.1</version> <!-- {x-version-update;com.azure:azure-core;dependency} -->
+      </dependency>
     </dependencies>
   </dependencyManagement>
 

sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/main/java/com/azure/spring/sample/eventhubs/kafka/EventHubKafkaBinderApplication.java

@@ -3,16 +3,44 @@
 
 package com.azure.spring.sample.eventhubs.kafka;
 
+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 reactor.core.publisher.Flux;
+import reactor.core.publisher.Sinks;
+
+import java.util.function.Consumer;
+import java.util.function.Supplier;
 
 /**
  * @author Warren Zhu
  */
 @SpringBootApplication
 public class EventHubKafkaBinderApplication {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(EventHubKafkaBinderApplication.class);
+
     public static void main(String[] args) {
         SpringApplication.run(EventHubKafkaBinderApplication.class, args);
     }
+
+    @Bean
+    public Sinks.Many<Message<String>> many() {
+        return Sinks.many().unicast().onBackpressureBuffer();
+    }
+
+    @Bean
+    public Supplier<Flux<Message<String>>> supply(Sinks.Many<Message<String>> many) {
+        return () -> many.asFlux()
+                         .doOnNext(m -> LOGGER.info("Manually sending message {}", m))
+                         .doOnError(t -> LOGGER.error("Error encountered", t));
+    }
+
+    @Bean
+    public Consumer<Message<String>> consume() {
+        return message -> LOGGER.info("New message received: '{}'", message.getPayload());
+    }
 }

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

@@ -4,26 +4,25 @@
 package com.azure.spring.sample.eventhubs.kafka;
 
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.cloud.stream.annotation.EnableBinding;
-import org.springframework.cloud.stream.messaging.Source;
+import org.springframework.messaging.Message;
 import org.springframework.messaging.support.GenericMessage;
 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.Sinks;
 
 /**
  * @author Warren Zhu
  */
-@EnableBinding(Source.class)
 @RestController
 public class SourceExample {
 
     @Autowired
-    private Source source;
+    private Sinks.Many<Message<String>> many;
 
     @PostMapping("/messages")
     public String sendMessage(@RequestParam String message) {
-        this.source.output().send(new GenericMessage<>(message));
+        many.emitNext(new GenericMessage<>(message), Sinks.EmitFailureHandler.FAIL_FAST);
         return message;
     }
 }

sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/main/resources/application-autocreate.yaml

@@ -11,9 +11,11 @@ spring:
       eventhub:
         namespace: [eventhub-namespace]
     stream:
+      function:
+        definition: consume;supply
       bindings:
-        input:
+        consume-in-0:
           destination: [eventhub-name]
           group: [consumer-group]
-        output:
-          destination: [the-same-eventhub-name-as-above]
+        supply-out-0:
+          destination: [the-same-eventhub-name-as-above]

sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/main/resources/application.yaml

@@ -8,9 +8,11 @@ spring:
       eventhub:
         namespace: [eventhub-namespace]
     stream:
+      function:
+        definition: consume;supply
       bindings:
-        input:
+        consume-in-0:
           destination: [eventhub-name]
           group: [consumer-group]
-        output:
+        supply-out-0:
           destination: [the-same-eventhub-name-as-above]