diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-boot-sample-servicebus/src/main/java/com/azure/spring/sample/servicebus/ServiceBusSampleApplication.java b/sdk/spring/azure-spring-boot-samples/azure-spring-boot-sample-servicebus/src/main/java/com/azure/spring/sample/servicebus/ServiceBusSampleApplication.java index 8089b7f46491..115c0152a000 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-boot-sample-servicebus/src/main/java/com/azure/spring/sample/servicebus/ServiceBusSampleApplication.java +++ b/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); diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/README.md b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/README.md index ecd667fc4211..b5ec368ce223 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/README.md +++ b/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] diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/pom.xml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/pom.xml index df14c7212c24..16f3a4cffc00 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/pom.xml +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/pom.xml @@ -21,7 +21,12 @@ 2020.0.1 pom import - + + + com.azure + azure-core + 1.14.1 + diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/main/java/com/azure/spring/sample/eventhubs/binder/EventHubBinderApplication.java b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/main/java/com/azure/spring/sample/eventhubs/binder/EventHubBinderApplication.java index aceebe54c28c..5adb4e1ad25b 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/main/java/com/azure/spring/sample/eventhubs/binder/EventHubBinderApplication.java +++ b/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> 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(); }; } diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/main/java/com/azure/spring/sample/eventhubs/binder/EventProducerController.java b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/main/java/com/azure/spring/sample/eventhubs/binder/EventProducerController.java index 9e37141fc3d2..0f4eaf40e83f 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/main/java/com/azure/spring/sample/eventhubs/binder/EventProducerController.java +++ b/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> emitterProcessor; + private Sinks.Many> many; @PostMapping("/messages") public ResponseEntity 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("/") diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/main/java/com/azure/spring/sample/eventhubs/binder/ManualEventProducerConfiguration.java b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/main/java/com/azure/spring/sample/eventhubs/binder/ManualEventProducerConfiguration.java index d07a6ae5fb0f..97135dfed9e5 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/main/java/com/azure/spring/sample/eventhubs/binder/ManualEventProducerConfiguration.java +++ b/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> emitter() { - return EmitterProcessor.create(); + public Sinks.Many> many() { + return Sinks.many().unicast().onBackpressureBuffer(); } @Bean - public Supplier>> supply(EmitterProcessor> emitter) { - return () -> Flux.from(emitter) + public Supplier>> supply(Sinks.Many> many) { + return () -> many.asFlux() .doOnNext(m -> LOGGER.info("Manually sending message {}", m)) .doOnError(t -> LOGGER.error("Error encountered", t)); } diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/main/resources/application-mi.yaml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/main/resources/application-mi.yaml index 7b4a1a591cd1..65cda53b488b 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/main/resources/application-mi.yaml +++ b/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 diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/main/resources/application-sp.yaml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/main/resources/application-sp.yaml index ea7157d0efa8..762e1788928e 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/main/resources/application-sp.yaml +++ b/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 diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/main/resources/application.yaml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/main/resources/application.yaml index 898cbaa95c39..b7a929cbd231 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-binder/src/main/resources/application.yaml +++ b/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 diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/README.md b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/README.md index 98c6ade650d4..bc06a04c6b68 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/README.md +++ b/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] ``` diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/pom.xml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/pom.xml index e6bd9edfe3bc..ac93865dc50c 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/pom.xml +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/pom.xml @@ -23,6 +23,11 @@ pom import + + com.azure + azure-core + 1.14.1 + diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/main/java/com/azure/spring/sample/eventhubs/kafka/EventHubKafkaBinderApplication.java b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/main/java/com/azure/spring/sample/eventhubs/kafka/EventHubKafkaBinderApplication.java index 0934e72660e6..cf9e965457e0 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/main/java/com/azure/spring/sample/eventhubs/kafka/EventHubKafkaBinderApplication.java +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/main/java/com/azure/spring/sample/eventhubs/kafka/EventHubKafkaBinderApplication.java @@ -3,8 +3,17 @@ 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 @@ -12,7 +21,26 @@ @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> many() { + return Sinks.many().unicast().onBackpressureBuffer(); + } + + @Bean + public Supplier>> supply(Sinks.Many> many) { + return () -> many.asFlux() + .doOnNext(m -> LOGGER.info("Manually sending message {}", m)) + .doOnError(t -> LOGGER.error("Error encountered", t)); + } + + @Bean + public Consumer> consume() { + return message -> LOGGER.info("New message received: '{}'", message.getPayload()); + } } diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/main/java/com/azure/spring/sample/eventhubs/kafka/SinkExample.java b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/main/java/com/azure/spring/sample/eventhubs/kafka/SinkExample.java deleted file mode 100644 index d5fdbd1dfed0..000000000000 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/main/java/com/azure/spring/sample/eventhubs/kafka/SinkExample.java +++ /dev/null @@ -1,20 +0,0 @@ -// Copyright (c) Microsoft Corporation. All rights reserved. -// Licensed under the MIT License. - -package com.azure.spring.sample.eventhubs.kafka; - -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.annotation.StreamListener; -import org.springframework.cloud.stream.messaging.Sink; - -/** - * @author Warren Zhu - */ -@EnableBinding(Sink.class) -public class SinkExample { - - @StreamListener(Sink.INPUT) - public void handleMessage(String message) { - System.out.println(String.format("New message received: '%s'", message)); - } -} diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/main/java/com/azure/spring/sample/eventhubs/kafka/SourceExample.java b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/main/java/com/azure/spring/sample/eventhubs/kafka/SourceExample.java index 724811d885f6..d3d81a458a69 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/main/java/com/azure/spring/sample/eventhubs/kafka/SourceExample.java +++ b/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> 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; } } diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/main/resources/application-autocreate.yaml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/main/resources/application-autocreate.yaml index 525d61a2b804..6ba61a952f1a 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/main/resources/application-autocreate.yaml +++ b/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] \ No newline at end of file + supply-out-0: + destination: [the-same-eventhub-name-as-above] diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/main/resources/application.yaml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/main/resources/application.yaml index 60649b5bc82c..e471a598fea7 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-kafka/src/main/resources/application.yaml +++ b/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] \ No newline at end of file diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/README.md b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/README.md index c0d29b0622cd..fe67aa1b90f8 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/README.md +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/README.md @@ -36,11 +36,13 @@ and bill at [this link][azure-account]. service principal or managed identity, update the `application-sp.yaml` or `application-mi.yaml` respectively. - ```yaml spring: cloud: stream: + # To specify which functional bean to bind to the external destination(s) exposed by the bindings + function: + definition: consume1;supply1;consume2;supply2 bindings: consume1-in-0: destination: [eventhub-1-name] @@ -88,14 +90,9 @@ and bill at [this link][azure-account]. consume2-in-0: consumer: checkpoint-mode: MANUAL - - #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: diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/pom.xml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/pom.xml index b7162704537b..5a0105888b9f 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/pom.xml +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/pom.xml @@ -24,6 +24,11 @@ pom import + + com.azure + azure-core + 1.14.1 + diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/java/com/azure/spring/sample/eventhubs/multibinders/EventHubMultiBindersApplication.java b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/java/com/azure/spring/sample/eventhubs/multibinders/EventHubMultiBindersApplication.java index 1e9c1c9f8668..e68005a2c56b 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/java/com/azure/spring/sample/eventhubs/multibinders/EventHubMultiBindersApplication.java +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/java/com/azure/spring/sample/eventhubs/multibinders/EventHubMultiBindersApplication.java @@ -31,7 +31,7 @@ public Consumer> consume1() { LOGGER.info("New message1 received: '{}'", message); checkpointer.success() .doOnSuccess(success -> LOGGER.info("Message1 '{}' successfully checkpointed", message)) - .doOnError(error -> LOGGER.error("Exception: {}", error.getMessage())) + .doOnError(error -> LOGGER.error("Exception found", error)) .subscribe(); }; } @@ -43,7 +43,7 @@ public Consumer> consume2() { LOGGER.info("New message2 received: '{}'", message); checkpointer.success() .doOnSuccess(success -> LOGGER.info("Message2 '{}' successfully checkpointed", message)) - .doOnError(error -> LOGGER.error("Exception: {}", error.getMessage())) + .doOnError(error -> LOGGER.error("Exception found", error)) .subscribe(); }; } diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/java/com/azure/spring/sample/eventhubs/multibinders/EventProducerController.java b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/java/com/azure/spring/sample/eventhubs/multibinders/EventProducerController.java index 60ef3352ba1e..ef3523b869d4 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/java/com/azure/spring/sample/eventhubs/multibinders/EventProducerController.java +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/java/com/azure/spring/sample/eventhubs/multibinders/EventProducerController.java @@ -12,33 +12,34 @@ 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; import javax.annotation.Resource; + @Profile("manual") @RestController public class EventProducerController { private static final Logger LOGGER = LoggerFactory.getLogger(EventHubMultiBindersApplication.class); - @Resource(name = "emitterProcessor1") - private EmitterProcessor> emitterProcessor1; + @Resource(name = "many1") + private Sinks.Many> many1; - @Resource(name = "emitterProcessor2") - private EmitterProcessor> emitterProcessor2; + @Resource(name = "many2") + private Sinks.Many> many2; @PostMapping("/messages1") public ResponseEntity sendMessage1(@RequestParam String message) { - LOGGER.info("Going to add message {} to emitter1", message); - emitterProcessor1.onNext(MessageBuilder.withPayload(message).build()); + LOGGER.info("Going to add message {} to Sinks.Many1", message); + many1.emitNext(MessageBuilder.withPayload(message).build(), Sinks.EmitFailureHandler.FAIL_FAST); return ResponseEntity.ok("Sent1!"); } @PostMapping("/messages2") public ResponseEntity sendMessage2(@RequestParam String message) { - LOGGER.info("Going to add message {} to emitter2", message); - emitterProcessor2.onNext(MessageBuilder.withPayload(message).build()); + LOGGER.info("Going to add message {} to Sinks.Many2.", message); + many2.emitNext(MessageBuilder.withPayload(message).build(), Sinks.EmitFailureHandler.FAIL_FAST); return ResponseEntity.ok("Sent2!"); } } diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/java/com/azure/spring/sample/eventhubs/multibinders/ManualEventProducerConfiguration.java b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/java/com/azure/spring/sample/eventhubs/multibinders/ManualEventProducerConfiguration.java index bfd52395a4d2..c5664888aef0 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/java/com/azure/spring/sample/eventhubs/multibinders/ManualEventProducerConfiguration.java +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/java/com/azure/spring/sample/eventhubs/multibinders/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,27 +21,27 @@ public class ManualEventProducerConfiguration { private static final Logger LOGGER = LoggerFactory.getLogger(EventHubMultiBindersApplication.class); @Bean - public EmitterProcessor> emitterProcessor1() { - return EmitterProcessor.create(); + public Sinks.Many> many1() { + return Sinks.many().unicast().onBackpressureBuffer(); } @Bean - public EmitterProcessor> emitterProcessor2() { - return EmitterProcessor.create(); + public Sinks.Many> many2() { + return Sinks.many().unicast().onBackpressureBuffer(); } @Bean - public Supplier>> supply1(EmitterProcessor> emitterProcessor1) { - return () -> Flux.from(emitterProcessor1) - .doOnNext(m -> LOGGER.info("Manually sending message1 {}", m)) - .doOnError(t -> LOGGER.error("Error encountered", t)); + public Supplier>> supply1(Sinks.Many> many1) { + return () -> many1.asFlux() + .doOnNext(m -> LOGGER.info("Manually sending message1 {}", m)) + .doOnError(t -> LOGGER.error("Error encountered", t)); } @Bean - public Supplier>> supply2(EmitterProcessor> emitterProcessor2) { - return () -> Flux.from(emitterProcessor2) - .doOnNext(m -> LOGGER.info("Manually sending message2 {}", m)) - .doOnError(t -> LOGGER.error("Error encountered", t)); + public Supplier>> supply2(Sinks.Many> many2) { + return () -> many2.asFlux() + .doOnNext(m -> LOGGER.info("Manually sending message2 {}", m)) + .doOnError(t -> LOGGER.error("Error encountered", t)); } } diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/resources/application-mi.yaml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/resources/application-mi.yaml index 8dacf2980d60..f1bc5fbabd8b 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/resources/application-mi.yaml +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/resources/application-mi.yaml @@ -10,6 +10,8 @@ spring: # auto-create-resources: true # region: [region] stream: + function: + definition: consume1;supply1;consume2;supply2 bindings: consume1-in-0: destination: [eventhub-1-name] @@ -55,10 +57,6 @@ spring: consume2-in-0: consumer: checkpoint-mode: MANUAL - - function: - definition: consume1;supply1;consume2;supply2; poller: initial-delay: 0 fixed-delay: 1000 - diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/resources/application-sp.yaml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/resources/application-sp.yaml index 42cbfefbabec..d262db1d80b6 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/resources/application-sp.yaml +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/resources/application-sp.yaml @@ -11,6 +11,8 @@ spring: # region: [region] # subscription-id: [subscription-id] stream: + function: + definition: consume1;supply1;consume2;supply2 bindings: consume1-in-0: destination: [eventhub-1-name] @@ -56,9 +58,6 @@ spring: consume2-in-0: consumer: checkpoint-mode: MANUAL - - function: - definition: consume1;supply1;consume2;supply2; poller: initial-delay: 0 fixed-delay: 1000 diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/resources/application.yaml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/resources/application.yaml index 7806c76a3c18..b0ab1a28e13b 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/resources/application.yaml +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-multibinders/src/main/resources/application.yaml @@ -1,6 +1,8 @@ spring: cloud: stream: + function: + definition: consume1;supply1;consume2;supply2 bindings: consume1-in-0: destination: [eventhub-1-name] @@ -48,10 +50,6 @@ spring: consume2-in-0: consumer: checkpoint-mode: MANUAL - - function: - definition: consume1;supply1;consume2;supply2; poller: initial-delay: 0 fixed-delay: 1000 - diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-operation/pom.xml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-operation/pom.xml index 9103d6bf284e..85cfe867e12c 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-operation/pom.xml +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-operation/pom.xml @@ -23,6 +23,11 @@ pom import + + com.azure + azure-core + 1.14.1 + diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-operation/src/main/java/com/azure/spring/sample/eventhubs/operation/WebController.java b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-operation/src/main/java/com/azure/spring/sample/eventhubs/operation/WebController.java index ecce96ae8de7..11c5e33dd712 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-operation/src/main/java/com/azure/spring/sample/eventhubs/operation/WebController.java +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-eventhubs-operation/src/main/java/com/azure/spring/sample/eventhubs/operation/WebController.java @@ -8,6 +8,8 @@ import com.azure.spring.integration.core.api.CheckpointMode; import com.azure.spring.integration.core.api.reactor.Checkpointer; import com.azure.spring.integration.eventhub.api.EventHubOperation; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; @@ -23,6 +25,8 @@ @RestController public class WebController { + private static final Logger LOGGER = LoggerFactory.getLogger(WebController.class); + private static final String EVENT_HUB_NAME = "eventhub1"; private static final String CONSUMER_GROUP = "cg1"; @@ -43,12 +47,12 @@ public void subscribeToEventHub() { } private void messageReceiver(Message message) { - System.out.println(String.format("New message received: '%s'", message.getPayload())); + LOGGER.info("New message received: '{}'", message.getPayload()); Checkpointer checkpointer = message.getHeaders().get(AzureHeaders.CHECKPOINTER, Checkpointer.class); checkpointer.success() - .doOnSuccess(s -> System.out.println(String.format("Message '%s' successfully checkpointed", - message.getPayload()))) - .doOnError(System.out::println) + .doOnSuccess(s -> LOGGER.info("Message '{}' successfully checkpointed", + message.getPayload())) + .doOnError(e -> LOGGER.error("Error found", e)) .subscribe(); } } diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-messaging/pom.xml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-messaging/pom.xml index d59482251348..93b943976c1f 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-messaging/pom.xml +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-messaging/pom.xml @@ -23,6 +23,11 @@ pom import + + com.azure + azure-core + 1.14.1 + diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-messaging/src/main/java/com/azure/spring/sample/messaging/WebController.java b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-messaging/src/main/java/com/azure/spring/sample/messaging/WebController.java index 018a46320f07..145863a92018 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-messaging/src/main/java/com/azure/spring/sample/messaging/WebController.java +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-messaging/src/main/java/com/azure/spring/sample/messaging/WebController.java @@ -5,6 +5,8 @@ import com.azure.spring.integration.eventhub.api.EventHubOperation; import com.azure.spring.messaging.annotation.AzureMessageListener; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.support.MessageBuilder; import org.springframework.web.bind.annotation.PostMapping; @@ -17,6 +19,7 @@ @RestController public class WebController { + private static final Logger LOGGER = LoggerFactory.getLogger(WebController.class); private static final String EVENT_HUB_NAME = "event-hub-name"; private static final String CONSUMER_GROUP = "$Default"; @@ -31,6 +34,6 @@ public User send(@RequestBody User user) { @AzureMessageListener(destination = EVENT_HUB_NAME, group = CONSUMER_GROUP) public void handleMessage(User user) { - System.out.println(String.format("New message received: '%s'", user)); + LOGGER.info("New message received: '{}'", user); } } diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-operation/src/main/java/com/azure/spring/sample/servicebus/operation/QueueController.java b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-operation/src/main/java/com/azure/spring/sample/servicebus/operation/QueueController.java index 43a8ea650fac..742431224580 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-operation/src/main/java/com/azure/spring/sample/servicebus/operation/QueueController.java +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-operation/src/main/java/com/azure/spring/sample/servicebus/operation/QueueController.java @@ -8,6 +8,8 @@ import com.azure.spring.integration.core.api.CheckpointMode; import com.azure.spring.integration.core.api.Checkpointer; import com.azure.spring.integration.servicebus.queue.ServiceBusQueueOperation; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; @@ -23,6 +25,7 @@ @RestController public class QueueController { + private static final Logger LOGGER = LoggerFactory.getLogger(QueueController.class); private static final String QUEUE_NAME = "queue1"; @Autowired @@ -42,11 +45,11 @@ public void subscribe() { } private void messageReceiver(Message message) { - System.out.println(String.format("New message received: '%s'", message.getPayload())); + LOGGER.info("New message received: '{}'", message.getPayload()); Checkpointer checkpointer = message.getHeaders().get(AzureHeaders.CHECKPOINTER, Checkpointer.class); checkpointer.success().handle((r, ex) -> { if (ex == null) { - System.out.println(String.format("Message '%s' successfully checkpointed", message.getPayload())); + LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()); } return null; }); diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-operation/src/main/java/com/azure/spring/sample/servicebus/operation/TopicController.java b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-operation/src/main/java/com/azure/spring/sample/servicebus/operation/TopicController.java index 82e44d65018b..ca6384e36df4 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-operation/src/main/java/com/azure/spring/sample/servicebus/operation/TopicController.java +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-operation/src/main/java/com/azure/spring/sample/servicebus/operation/TopicController.java @@ -8,6 +8,8 @@ import com.azure.spring.integration.core.api.CheckpointMode; import com.azure.spring.integration.core.api.Checkpointer; import com.azure.spring.integration.servicebus.topic.ServiceBusTopicOperation; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; @@ -23,6 +25,7 @@ @RestController public class TopicController { + private static final Logger LOGGER = LoggerFactory.getLogger(TopicController.class); private static final String TOPIC_NAME = "topic1"; private static final String SUBSCRIPTION_NAME = "group1"; @@ -43,11 +46,11 @@ public void subscribe() { } private void messageReceiver(Message message) { - System.out.println(String.format("New message received: '%s'", message.getPayload())); + LOGGER.info("New message received: '{}'", message.getPayload()); Checkpointer checkpointer = message.getHeaders().get(AzureHeaders.CHECKPOINTER, Checkpointer.class); checkpointer.success().handle((r, ex) -> { if (ex == null) { - System.out.println(String.format("Message '%s' successfully checkpointed", message.getPayload())); + LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()); } return null; }); diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-binder/README.md b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-binder/README.md index 28054203a486..a1b224e5ba30 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-binder/README.md +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-binder/README.md @@ -37,13 +37,13 @@ Bus Queue. You can choose anyone of them. servicebus: connection-string: [servicebus-namespace-connection-string] stream: + function: + definition: consume;supply bindings: consume-in-0: destination: [servicebus-queue-name] supply-out-0: destination: [servicebus-queue-name-same-as-above] - function: - definition: consume;supply; poller: fixed-delay: 1000 initial-delay: 0 @@ -73,13 +73,13 @@ Bus Queue. You can choose anyone of them. servicebus: namespace: [servicebus-namespace] stream: + function: + definition: consume;supply bindings: consume-in-0: destination: [servicebus-queue-name] supply-out-0: destination: [servicebus-queue-name-same-as-above] - function: - definition: consume;supply; poller: fixed-delay: 1000 initial-delay: 0 @@ -114,13 +114,13 @@ Please follow [create managed identity][create-managed-identity] to set up manag servicebus: namespace: [servicebus-namespace] stream: + function: + definition: consume;supply bindings: consume-in-0: destination: [servicebus-queue-name] supply-out-0: destination: [servicebus-queue-name-same-as-above] - function: - definition: consume;supply; poller: fixed-delay: 1000 initial-delay: 0 diff --git a/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 b/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 index 18368747ad49..62597bf371fa 100644 --- a/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 +++ b/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 @@ -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 ManualServiceProducerConfiguration { private static final Logger LOGGER = LoggerFactory.getLogger(ServiceBusQueueBinderApplication.class); @Bean - public EmitterProcessor> emitter() { - return EmitterProcessor.create(); + public Sinks.Many> many() { + return Sinks.many().unicast().onBackpressureBuffer(); } @Bean - public Supplier>> supply(EmitterProcessor> emitter) { - return () -> Flux.from(emitter) + public Supplier>> supply(Sinks.Many> many) { + return () -> many.asFlux() .doOnNext(m -> LOGGER.info("Manually sending message {}", m)) .doOnError(t -> LOGGER.error("Error encountered", t)); } diff --git a/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 b/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 index 7012147e814c..9233be6252dc 100644 --- a/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 +++ b/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 @@ -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; @RestController @Profile("manual") @@ -23,12 +23,12 @@ public class ServiceProducerController { private static final Logger LOGGER = LoggerFactory.getLogger(ServiceBusQueueBinderApplication.class); @Autowired - private EmitterProcessor> emitterProcessor; + private Sinks.Many> many; @PostMapping("/messages") public ResponseEntity sendMessage(@RequestParam String message) { - LOGGER.info("Going to add message {} to emitter", message); - emitterProcessor.onNext(MessageBuilder.withPayload(message).build()); + LOGGER.info("Going to add message {} to Sinks.Many.", message); + many.emitNext(MessageBuilder.withPayload(message).build(), Sinks.EmitFailureHandler.FAIL_FAST); return ResponseEntity.ok("Sent!"); } diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-binder/src/main/resources/application-mi.yaml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-binder/src/main/resources/application-mi.yaml index 17b12089343a..cb21f97dfbb7 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-binder/src/main/resources/application-mi.yaml +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-binder/src/main/resources/application-mi.yaml @@ -15,6 +15,8 @@ spring: servicebus: namespace: [servicebus-namespace] stream: + function: + definition: consume;supply bindings: consume-in-0: destination: [servicebus-queue-name] @@ -26,8 +28,6 @@ spring: consume-in-0: consumer: checkpoint-mode: MANUAL - function: - definition: consume;supply; poller: fixed-delay: 1000 initial-delay: 0 diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-binder/src/main/resources/application-sp.yaml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-binder/src/main/resources/application-sp.yaml index 9c886df7e7e0..a8b4a646d8d1 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-binder/src/main/resources/application-sp.yaml +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-binder/src/main/resources/application-sp.yaml @@ -16,6 +16,8 @@ spring: servicebus: namespace: [servicebus-namespace] stream: + function: + definition: consume;supply bindings: consume-in-0: destination: [servicebus-queue-name] @@ -27,8 +29,6 @@ spring: consume-in-0: consumer: checkpoint-mode: MANUAL - function: - definition: consume;supply; poller: fixed-delay: 1000 initial-delay: 0 diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-binder/src/main/resources/application.yaml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-binder/src/main/resources/application.yaml index fede84c765e0..aa7ea22ccff8 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-binder/src/main/resources/application.yaml +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-binder/src/main/resources/application.yaml @@ -4,6 +4,8 @@ spring: servicebus: connection-string: [servicebus-namespace-connection-string] stream: + function: + definition: consume;supply bindings: consume-in-0: destination: [servicebus-queue-name] @@ -15,8 +17,6 @@ spring: consume-in-0: consumer: checkpoint-mode: MANUAL - function: - definition: consume;supply; poller: fixed-delay: 1000 initial-delay: 0 diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/README.md b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/README.md index c910ed309e23..76006e74d305 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/README.md +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/README.md @@ -37,6 +37,9 @@ and bill at [this link][azure-account]. spring: cloud: stream: + #To specify which functional bean to bind to the external destination(s) exposed by the bindings + function: + definition: consume1;supply1;consume2;supply2 bindings: consume1-in-0: destination: [servicebus-queue-1-name] @@ -68,10 +71,6 @@ and bill at [this link][azure-account]. azure: servicebus: 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 diff --git a/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 b/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 index afb6e9d6166a..0606b8ccbfeb 100644 --- a/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 +++ b/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 @@ -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,26 +21,26 @@ public class ManualServiceProducerConfiguration { private static final Logger LOGGER = LoggerFactory.getLogger(ServiceBusQueueMultiBindersApplication.class); @Bean - public EmitterProcessor> emitterProcessor1() { - return EmitterProcessor.create(); + public Sinks.Many> many1() { + return Sinks.many().unicast().onBackpressureBuffer(); } @Bean - public EmitterProcessor> emitterProcessor2() { - return EmitterProcessor.create(); + public Sinks.Many> many2() { + return Sinks.many().unicast().onBackpressureBuffer(); } @Bean - public Supplier>> supply1(EmitterProcessor> emitterProcessor1) { - return () -> Flux.from(emitterProcessor1) - .doOnNext(m -> LOGGER.info("Manually sending message1 {}", m)) - .doOnError(t -> LOGGER.error("Error encountered", t)); + public Supplier>> supply1(Sinks.Many> many1) { + return () -> many1.asFlux() + .doOnNext(m -> LOGGER.info("Manually sending message1 {}", m)) + .doOnError(t -> LOGGER.error("Error encountered", t)); } @Bean - public Supplier>> supply2(EmitterProcessor> emitterProcessor2) { - return () -> Flux.from(emitterProcessor2) - .doOnNext(m -> LOGGER.info("Manually sending message2 {}", m)) - .doOnError(t -> LOGGER.error("Error encountered", t)); + public Supplier>> supply2(Sinks.Many> many2) { + return () -> many2.asFlux() + .doOnNext(m -> LOGGER.info("Manually sending message2 {}", m)) + .doOnError(t -> LOGGER.error("Error encountered", t)); } } diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/src/main/java/com/azure/spring/sample/servicebus/queue/multibinders/ServiceProducerController.java b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/src/main/java/com/azure/spring/sample/servicebus/queue/multibinders/ServiceProducerController.java index 0a5cce716c8b..baaccf32121c 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/src/main/java/com/azure/spring/sample/servicebus/queue/multibinders/ServiceProducerController.java +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/src/main/java/com/azure/spring/sample/servicebus/queue/multibinders/ServiceProducerController.java @@ -12,7 +12,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; import javax.annotation.Resource; @@ -22,23 +22,23 @@ public class ServiceProducerController { private static final Logger LOGGER = LoggerFactory.getLogger(ServiceBusQueueMultiBindersApplication.class); - @Resource(name = "emitterProcessor1") - private EmitterProcessor> emitterProcessor1; + @Resource(name = "many1") + private Sinks.Many> many1; - @Resource(name = "emitterProcessor2") - private EmitterProcessor> emitterProcessor2; + @Resource(name = "many2") + private Sinks.Many> many2; @PostMapping("/messages1") public ResponseEntity sendMessage1(@RequestParam String message) { - LOGGER.info("Going to add message {} to emitter1", message); - emitterProcessor1.onNext(MessageBuilder.withPayload(message).build()); + LOGGER.info("Going to add message {} to Sinks.Many1.", message); + many1.emitNext(MessageBuilder.withPayload(message).build(), Sinks.EmitFailureHandler.FAIL_FAST); return ResponseEntity.ok("Sent1!"); } @PostMapping("/messages2") public ResponseEntity sendMessage2(@RequestParam String message) { - LOGGER.info("Going to add message {} to emitter2", message); - emitterProcessor2.onNext(MessageBuilder.withPayload(message).build()); + LOGGER.info("Going to add message {} to Sinks.Many2.", message); + many2.emitNext(MessageBuilder.withPayload(message).build(), Sinks.EmitFailureHandler.FAIL_FAST); return ResponseEntity.ok("Sent2!"); } } diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/src/main/resources/application-mi.yaml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/src/main/resources/application-mi.yaml index 68bda8ffddd2..4566ed39a9f0 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/src/main/resources/application-mi.yaml +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/src/main/resources/application-mi.yaml @@ -11,6 +11,8 @@ spring: # region: [region] stream: + function: + definition: consume1;supply1;consume2;supply2 bindings: consume1-in-0: destination: [servicebus-queue-1-name] @@ -51,8 +53,6 @@ spring: consume2-in-0: consumer: checkpoint-mode: MANUAL - function: - definition: consume1;supply1;consume2;supply2; poller: initial-delay: 0 fixed-delay: 1000 diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/src/main/resources/application-sp.yaml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/src/main/resources/application-sp.yaml index c24aac002a59..aae579b01ed4 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/src/main/resources/application-sp.yaml +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/src/main/resources/application-sp.yaml @@ -12,6 +12,8 @@ spring: # subscription-id: [subscription-id] stream: + function: + definition: consume1;supply1;consume2;supply2 bindings: consume1-in-0: destination: [servicebus-queue-1-name] @@ -52,8 +54,6 @@ spring: consume2-in-0: consumer: checkpoint-mode: MANUAL - function: - definition: consume1;supply1;consume2;supply2; poller: initial-delay: 0 fixed-delay: 1000 diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/src/main/resources/application.yaml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/src/main/resources/application.yaml index a01ab743f2fe..4699e03fa467 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/src/main/resources/application.yaml +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-queue-multibinders/src/main/resources/application.yaml @@ -1,6 +1,8 @@ spring: cloud: stream: + function: + definition: consume1;supply1;consume2;supply2 bindings: consume1-in-0: destination: [servicebus-queue-1-name] @@ -41,8 +43,6 @@ spring: consume2-in-0: consumer: checkpoint-mode: MANUAL - function: - definition: consume1;supply1;consume2;supply2; poller: initial-delay: 0 fixed-delay: 1000 diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/README.md b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/README.md index 5d1ac8f917f4..1a99020f73d7 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/README.md +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/README.md @@ -36,13 +36,13 @@ Service Bus Topic. You can choose anyone of them. servicebus: connection-string: [servicebus-namespace-connection-string] stream: + function: + definition: consume;supply bindings: consume-in-0: destination: [servicebus-queue-name] supply-out-0: destination: [servicebus-queue-name-same-as-above] - function: - definition: consume;supply; poller: fixed-delay: 1000 initial-delay: 0 @@ -72,13 +72,13 @@ Service Bus Topic. You can choose anyone of them. servicebus: namespace: [servicebus-namespace] stream: + function: + definition: consume;supply bindings: consume-in-0: destination: [servicebus-queue-name] supply-out-0: destination: [servicebus-queue-name-same-as-above] - function: - definition: consume;supply; poller: fixed-delay: 1000 initial-delay: 0 @@ -115,13 +115,13 @@ Please follow [create managed identity][create-managed-identity] to set up manag servicebus: namespace: [servicebus-namespace] stream: + function: + definition: consume;supply bindings: consume-in-0: destination: [servicebus-queue-name] supply-out-0: destination: [servicebus-queue-name-same-as-above] - function: - definition: consume;supply; poller: fixed-delay: 1000 initial-delay: 0 diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/src/main/java/com/azure/spring/sample/servicebus/topic/binder/ManualServiceProducerConfiguration.java b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/src/main/java/com/azure/spring/sample/servicebus/topic/binder/ManualServiceProducerConfiguration.java index ff4aeff5b74b..d2a04f0cbc2f 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/src/main/java/com/azure/spring/sample/servicebus/topic/binder/ManualServiceProducerConfiguration.java +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/src/main/java/com/azure/spring/sample/servicebus/topic/binder/ManualServiceProducerConfiguration.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 ManualServiceProducerConfiguration { private static final Logger LOGGER = LoggerFactory.getLogger(ServiceBusTopicBinderApplication.class); @Bean - public EmitterProcessor> emitter() { - return EmitterProcessor.create(); + public Sinks.Many> many() { + return Sinks.many().unicast().onBackpressureBuffer(); } @Bean - public Supplier>> supply(EmitterProcessor> emitter) { - return () -> Flux.from(emitter) + public Supplier>> supply(Sinks.Many> many) { + return () -> many.asFlux() .doOnNext(m -> LOGGER.info("Manually sending message {}", m)) .doOnError(t -> LOGGER.error("Error encountered", t)); } diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/src/main/java/com/azure/spring/sample/servicebus/topic/binder/ServiceProducerController.java b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/src/main/java/com/azure/spring/sample/servicebus/topic/binder/ServiceProducerController.java index d00ed73339bb..10575a319b09 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/src/main/java/com/azure/spring/sample/servicebus/topic/binder/ServiceProducerController.java +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/src/main/java/com/azure/spring/sample/servicebus/topic/binder/ServiceProducerController.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; @RestController @Profile("manual") @@ -23,12 +23,12 @@ public class ServiceProducerController { private static final Logger LOGGER = LoggerFactory.getLogger(ServiceBusTopicBinderApplication.class); @Autowired - private EmitterProcessor> emitterProcessor; + private Sinks.Many> many; @PostMapping("/messages") public ResponseEntity sendMessage(@RequestParam String message) { - LOGGER.info("Going to add message {} to emitter", message); - emitterProcessor.onNext(MessageBuilder.withPayload(message).build()); + LOGGER.info("Going to add message {} to Sinks.Many.", message); + many.emitNext(MessageBuilder.withPayload(message).build(), Sinks.EmitFailureHandler.FAIL_FAST); return ResponseEntity.ok("Sent!"); } diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/src/main/resources/application-mi.yaml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/src/main/resources/application-mi.yaml index 2ebadc191401..df62f5a127ce 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/src/main/resources/application-mi.yaml +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/src/main/resources/application-mi.yaml @@ -15,6 +15,8 @@ spring: servicebus: namespace: [servicebus-namespace] stream: + function: + definition: consume;supply bindings: consume-in-0: destination: [servicebus-queue-name] @@ -26,8 +28,6 @@ spring: consume-in-0: consumer: checkpoint-mode: MANUAL - function: - definition: consume;supply; poller: fixed-delay: 1000 initial-delay: 0 diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/src/main/resources/application-sp.yaml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/src/main/resources/application-sp.yaml index 35ab334091e5..dece30b0ee07 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/src/main/resources/application-sp.yaml +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/src/main/resources/application-sp.yaml @@ -16,6 +16,8 @@ spring: servicebus: namespace: [servicebus-namespace] stream: + function: + definition: consume;supply bindings: consume-in-0: destination: [servicebus-topic-name] @@ -28,8 +30,6 @@ spring: consume-in-0: consumer: checkpoint-mode: MANUAL - function: - definition: consume;supply; poller: fixed-delay: 1000 initial-delay: 0 diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/src/main/resources/application.yaml b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/src/main/resources/application.yaml index df4e686777d0..204339b8adcd 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/src/main/resources/application.yaml +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-cloud-sample-servicebus-topic-binder/src/main/resources/application.yaml @@ -4,6 +4,8 @@ spring: servicebus: connection-string: [servicebus-namespace-connection-string] stream: + function: + definition: consume;supply bindings: consume-in-0: destination: [servicebus-topic-name] @@ -16,8 +18,6 @@ spring: consume-in-0: consumer: checkpoint-mode: MANUAL - function: - definition: consume;supply; poller: fixed-delay: 1000 initial-delay: 0 diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-eventhubs/pom.xml b/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-eventhubs/pom.xml index e14ba33d1f3c..af94b34402ba 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-eventhubs/pom.xml +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-eventhubs/pom.xml @@ -23,6 +23,11 @@ pom import + + com.azure + azure-core + 1.14.1 + diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-eventhubs/src/main/java/com/azure/spring/sample/eventhubs/ReceiveController.java b/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-eventhubs/src/main/java/com/azure/spring/sample/eventhubs/ReceiveController.java index e6b3dac4bfde..df0b8603ba83 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-eventhubs/src/main/java/com/azure/spring/sample/eventhubs/ReceiveController.java +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-eventhubs/src/main/java/com/azure/spring/sample/eventhubs/ReceiveController.java @@ -9,6 +9,8 @@ import com.azure.spring.integration.core.api.reactor.Checkpointer; import com.azure.spring.integration.eventhub.api.EventHubOperation; import com.azure.spring.integration.eventhub.inbound.EventHubInboundChannelAdapter; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; import org.springframework.integration.annotation.ServiceActivator; @@ -23,6 +25,7 @@ @RestController public class ReceiveController { + private static final Logger LOGGER = LoggerFactory.getLogger(ReceiveController.class); private static final String INPUT_CHANNEL = "input"; private static final String EVENTHUB_NAME = "eventhub1"; private static final String CONSUMER_GROUP = "cg1"; @@ -34,10 +37,10 @@ public class ReceiveController { @ServiceActivator(inputChannel = INPUT_CHANNEL) public void messageReceiver(byte[] payload, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer) { String message = new String(payload); - System.out.println(String.format("New message received: '%s'", message)); + LOGGER.info("New message received: '{}'", message); checkpointer.success() - .doOnSuccess(s -> System.out.println(String.format("Message '%s' successfully checkpointed", message))) - .doOnError(System.out::println) + .doOnSuccess(s -> LOGGER.info("Message '{}' successfully checkpointed", message)) + .doOnError(e -> LOGGER.error("Error found", e)) .subscribe(); } diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-eventhubs/src/main/java/com/azure/spring/sample/eventhubs/SendController.java b/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-eventhubs/src/main/java/com/azure/spring/sample/eventhubs/SendController.java index cf800aa44354..de29471f1d9f 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-eventhubs/src/main/java/com/azure/spring/sample/eventhubs/SendController.java +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-eventhubs/src/main/java/com/azure/spring/sample/eventhubs/SendController.java @@ -5,6 +5,8 @@ import com.azure.spring.integration.core.api.reactor.DefaultMessageHandler; import com.azure.spring.integration.eventhub.api.EventHubOperation; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.integration.annotation.MessagingGateway; @@ -22,6 +24,7 @@ @RestController public class SendController { + private static final Logger LOGGER = LoggerFactory.getLogger(SendController.class); private static final String OUTPUT_CHANNEL = "output"; private static final String EVENTHUB_NAME = "eventhub1"; @@ -44,12 +47,12 @@ public MessageHandler messageSender(EventHubOperation queueOperation) { handler.setSendCallback(new ListenableFutureCallback() { @Override public void onSuccess(Void result) { - System.out.println("Message was sent successfully."); + LOGGER.info("Message was sent successfully."); } @Override public void onFailure(Throwable ex) { - System.out.println("There was an error sending the message."); + LOGGER.error("There was an error sending the message.", ex); } }); diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-servicebus/src/main/java/com/azure/spring/sample/servicebus/QueueReceiveController.java b/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-servicebus/src/main/java/com/azure/spring/sample/servicebus/QueueReceiveController.java index 17c615dd28e6..a0b86530b1b7 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-servicebus/src/main/java/com/azure/spring/sample/servicebus/QueueReceiveController.java +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-servicebus/src/main/java/com/azure/spring/sample/servicebus/QueueReceiveController.java @@ -9,6 +9,8 @@ import com.azure.spring.integration.core.api.Checkpointer; import com.azure.spring.integration.servicebus.inbound.ServiceBusQueueInboundChannelAdapter; import com.azure.spring.integration.servicebus.queue.ServiceBusQueueOperation; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; import org.springframework.integration.annotation.ServiceActivator; @@ -23,6 +25,7 @@ @RestController public class QueueReceiveController { + private static final Logger LOGGER = LoggerFactory.getLogger(QueueReceiveController.class); private static final String INPUT_CHANNEL = "queue.input"; private static final String QUEUE_NAME = "queue1"; @@ -33,10 +36,10 @@ public class QueueReceiveController { @ServiceActivator(inputChannel = INPUT_CHANNEL) public void messageReceiver(byte[] payload, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer) { String message = new String(payload); - System.out.printf("New message received: '%s'%n", message); + LOGGER.info("New message received: '{}'", message); checkpointer.success().handle((r, ex) -> { if (ex == null) { - System.out.printf("Message '%s' successfully checkpointed.%n", message); + LOGGER.info("Message '{}' successfully checkpointed.", message); } return null; }); diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-servicebus/src/main/java/com/azure/spring/sample/servicebus/QueueSendController.java b/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-servicebus/src/main/java/com/azure/spring/sample/servicebus/QueueSendController.java index faa7c5779d40..70edcbdf9d3a 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-servicebus/src/main/java/com/azure/spring/sample/servicebus/QueueSendController.java +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-servicebus/src/main/java/com/azure/spring/sample/servicebus/QueueSendController.java @@ -5,8 +5,8 @@ import com.azure.spring.integration.core.DefaultMessageHandler; import com.azure.spring.integration.servicebus.queue.ServiceBusQueueOperation; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.integration.annotation.MessagingGateway; @@ -24,7 +24,7 @@ @RestController public class QueueSendController { - private static final Log LOGGER = LogFactory.getLog(QueueSendController.class); + private static final Logger LOGGER = LoggerFactory.getLogger(QueueSendController.class); private static final String OUTPUT_CHANNEL = "queue.output"; private static final String QUEUE_NAME = "queue1"; diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-servicebus/src/main/java/com/azure/spring/sample/servicebus/TopicReceiveController.java b/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-servicebus/src/main/java/com/azure/spring/sample/servicebus/TopicReceiveController.java index c8febc8b3888..be6904c0a678 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-servicebus/src/main/java/com/azure/spring/sample/servicebus/TopicReceiveController.java +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-servicebus/src/main/java/com/azure/spring/sample/servicebus/TopicReceiveController.java @@ -9,6 +9,8 @@ import com.azure.spring.integration.core.api.Checkpointer; import com.azure.spring.integration.servicebus.inbound.ServiceBusTopicInboundChannelAdapter; import com.azure.spring.integration.servicebus.topic.ServiceBusTopicOperation; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; import org.springframework.integration.annotation.ServiceActivator; @@ -23,6 +25,7 @@ @RestController public class TopicReceiveController { + private static final Logger LOGGER = LoggerFactory.getLogger(TopicReceiveController.class); private static final String INPUT_CHANNEL = "topic.input"; private static final String TOPIC_NAME = "topic1"; private static final String SUBSCRIPTION_NAME = "group1"; @@ -34,10 +37,10 @@ public class TopicReceiveController { @ServiceActivator(inputChannel = INPUT_CHANNEL) public void messageReceiver(byte[] payload, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer) { String message = new String(payload); - System.out.println(String.format("New message received: '%s'", message)); + LOGGER.info("New message received: '{}'", message); checkpointer.success().handle((r, ex) -> { if (ex == null) { - System.out.println(String.format("Message '%s' successfully checkpointed", message)); + LOGGER.info("Message '{}' successfully checkpointed", message); } return null; }); diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-servicebus/src/main/java/com/azure/spring/sample/servicebus/TopicSendController.java b/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-servicebus/src/main/java/com/azure/spring/sample/servicebus/TopicSendController.java index 10b2afb1154e..886bc936a0a2 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-servicebus/src/main/java/com/azure/spring/sample/servicebus/TopicSendController.java +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-servicebus/src/main/java/com/azure/spring/sample/servicebus/TopicSendController.java @@ -23,7 +23,8 @@ */ @RestController public class TopicSendController { - private static final Logger log = LoggerFactory.getLogger(TopicSendController.class); + + private static final Logger LOGGER = LoggerFactory.getLogger(TopicSendController.class); private static final String OUTPUT_CHANNEL = "topic.output"; private static final String TOPIC_NAME = "topic1"; @@ -46,12 +47,12 @@ public MessageHandler topicMessageSender(ServiceBusTopicOperation topicOperation handler.setSendCallback(new ListenableFutureCallback() { @Override public void onSuccess(Void result) { - log.info("Message was sent successfully."); + LOGGER.info("Message was sent successfully."); } @Override public void onFailure(Throwable ex) { - log.info("There was an error sending the message."); + LOGGER.info("There was an error sending the message."); } }); diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-storage-queue/src/main/java/com/azure/spring/sample/storage/queue/ReceiveController.java b/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-storage-queue/src/main/java/com/azure/spring/sample/storage/queue/ReceiveController.java index b1b8dad1e847..54c0dfebed37 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-storage-queue/src/main/java/com/azure/spring/sample/storage/queue/ReceiveController.java +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-storage-queue/src/main/java/com/azure/spring/sample/storage/queue/ReceiveController.java @@ -8,6 +8,8 @@ import com.azure.spring.integration.core.api.reactor.Checkpointer; import com.azure.spring.integration.storage.queue.StorageQueueOperation; import com.azure.spring.integration.storage.queue.inbound.StorageQueueMessageSource; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.context.annotation.Bean; import org.springframework.integration.annotation.InboundChannelAdapter; import org.springframework.integration.annotation.Poller; @@ -21,6 +23,8 @@ */ @RestController public class ReceiveController { + + private static final Logger LOGGER = LoggerFactory.getLogger(ReceiveController.class); /*Storage queue name can only be made up of lowercase letters, the numbers and the hyphen(-).*/ private static final String STORAGE_QUEUE_NAME = "example"; private static final String INPUT_CHANNEL = "input"; @@ -41,10 +45,10 @@ public StorageQueueMessageSource storageQueueMessageSource(StorageQueueOperation @ServiceActivator(inputChannel = INPUT_CHANNEL) public void messageReceiver(byte[] payload, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer) { String message = new String(payload); - System.out.printf("New message received: '%s'%n", message); + LOGGER.info("New message received: '{}'", message); checkpointer.success() .doOnError(Throwable::printStackTrace) - .doOnSuccess(t -> System.out.printf("Message '%s' successfully checkpointed%n", message)) + .doOnSuccess(t -> LOGGER.info("Message '{}' successfully checkpointed", message)) .subscribe(); } diff --git a/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-storage-queue/src/main/java/com/azure/spring/sample/storage/queue/SendController.java b/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-storage-queue/src/main/java/com/azure/spring/sample/storage/queue/SendController.java index 0eeaf1e57fea..35fbc9453b9f 100644 --- a/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-storage-queue/src/main/java/com/azure/spring/sample/storage/queue/SendController.java +++ b/sdk/spring/azure-spring-boot-samples/azure-spring-integration-sample-storage-queue/src/main/java/com/azure/spring/sample/storage/queue/SendController.java @@ -5,8 +5,8 @@ import com.azure.spring.integration.core.api.reactor.DefaultMessageHandler; import com.azure.spring.integration.storage.queue.StorageQueueOperation; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.integration.annotation.MessagingGateway; @@ -23,10 +23,11 @@ */ @RestController public class SendController { + + private static final Logger LOGGER = LoggerFactory.getLogger(SendController.class); /*Storage queue name can only be made up of lowercase letters, the numbers and the hyphen(-).*/ private static final String STORAGE_QUEUE_NAME = "example"; private static final String OUTPUT_CHANNEL = "output"; - private static final Log LOGGER = LogFactory.getLog(SendController.class); @Autowired StorageQueueOutboundGateway storageQueueOutboundGateway; diff --git a/sdk/spring/azure-spring-boot-starter-active-directory-b2c/README.md b/sdk/spring/azure-spring-boot-starter-active-directory-b2c/README.md index 5926730fb62d..b24d6dff5fec 100644 --- a/sdk/spring/azure-spring-boot-starter-active-directory-b2c/README.md +++ b/sdk/spring/azure-spring-boot-starter-active-directory-b2c/README.md @@ -196,7 +196,7 @@ you should be redirected to login page. 4. Click link with the login user flow, you should be redirected Azure AD B2C to start the authentication process. -4. After you have logged in successfully, you should see the sample `home page` from the browser. +5. After you have logged in successfully, you should see the sample `home page` from the browser. ## Troubleshooting ### Enable client logging diff --git a/sdk/spring/azure-spring-cloud-stream-binder-eventhubs/README.md b/sdk/spring/azure-spring-cloud-stream-binder-eventhubs/README.md index 89379964b279..853ee095cef2 100644 --- a/sdk/spring/azure-spring-cloud-stream-binder-eventhubs/README.md +++ b/sdk/spring/azure-spring-cloud-stream-binder-eventhubs/README.md @@ -126,7 +126,7 @@ this channel is open by default, you can handle the error message in this way: // Replace group with spring.cloud.stream.bindings.input.group @ServiceActivator(inputChannel = "{destination}.{group}.errors") public void consumerError(Message message) { - System.out.println("Handling customer ERROR: " + message); + LOGGER.error("Handling customer ERROR: " + message); } ``` @@ -142,7 +142,7 @@ you can handle the error message in this way: // Replace destination with spring.cloud.stream.bindings.output.destination @ServiceActivator(inputChannel = "{destination}.errors") public void producerError(Message message) { - System.out.println("Handling Producer ERROR: " + message); + LOGGER.error("Handling Producer ERROR: " + message); } ```