From aedfdd3ff9f821c69e9cad569b49d58e6b49607c Mon Sep 17 00:00:00 2001 From: Hemant Tanwar Date: Mon, 6 Jul 2020 12:15:38 -0700 Subject: [PATCH] Removing bridge API for reactor. --- .../ServiceBusReceiverAsyncClient.java | 28 -------------- ...ReceiverAsyncClientJavaDocCodeSamples.java | 30 --------------- ...BusReceiverAsyncClientIntegrationTest.java | 38 ------------------- 3 files changed, 96 deletions(-) diff --git a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClient.java b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClient.java index b75f2e01523f..5206e8ab6c0e 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClient.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/main/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClient.java @@ -56,10 +56,6 @@ * other terminal scenarios. See {@link #receiveMessages()} for more information.

* {@codesnippet com.azure.messaging.servicebus.servicebusasyncreceiverclient.receive#all} * - *

Receive a maximum number of messages or until max a Duration

- *

This receives at most 15 messages, or until a duration of 30 seconds elapses. Whichever occurs first.

- * {@codesnippet com.azure.messaging.servicebus.servicebusasyncreceiverclient.receive#int-duration} - * *

Receive messages in {@link ReceiveMode#RECEIVE_AND_DELETE} mode from Service Bus resource

* {@codesnippet com.azure.messaging.servicebus.servicebusasyncreceiverclient.receiveWithReceiveAndDeleteMode} * @@ -886,30 +882,6 @@ public Flux receiveMessages() { } } - /** - * Receives a bounded stream of {@link ServiceBusReceivedMessage messages} from the Service Bus entity. This stream - * receives either {@code maxNumberOfMessages} are received or the {@code maxWaitTime} has elapsed. - * - * @param maxNumberOfMessages Maximum number of messages to receive. - * @param maxWaitTime Maximum time to wait. - * - * @return A bounded {@link Flux} of messages. - * @throws NullPointerException if {@code maxWaitTime} is null. - * @throws IllegalArgumentException if {@code maxNumberOfMessages} is less than 1. {@code maxWaitTime} is zero - * or a negative duration. - */ - public Flux receiveMessages(int maxNumberOfMessages, Duration maxWaitTime) { - if (maxNumberOfMessages < 1) { - return fluxError(logger, new IllegalArgumentException("'maxNumberOfMessages' cannot be less than 1.")); - } else if (maxWaitTime == null) { - return fluxError(logger, new NullPointerException("'maxWaitTime' cannot be null.")); - } else if (maxWaitTime.isNegative() || maxWaitTime.isZero()) { - return fluxError(logger, new NullPointerException("'maxWaitTime' cannot be negative or zero.")); - } - - return receiveMessages().take(maxNumberOfMessages).take(maxWaitTime); - } - /** * Receives a deferred {@link ServiceBusReceivedMessage message}. Deferred messages can only be received by using * sequence number. diff --git a/sdk/servicebus/azure-messaging-servicebus/src/samples/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClientJavaDocCodeSamples.java b/sdk/servicebus/azure-messaging-servicebus/src/samples/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClientJavaDocCodeSamples.java index 7dde0409d038..4fa81a2c8c40 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/samples/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClientJavaDocCodeSamples.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/samples/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClientJavaDocCodeSamples.java @@ -11,7 +11,6 @@ import reactor.core.publisher.Mono; import java.util.Map; -import java.time.Duration; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -140,35 +139,6 @@ public void receiveAll() { // END: com.azure.messaging.servicebus.servicebusasyncreceiverclient.receive#all } - /** - * Receives messages up to a time or duration. - */ - public void receiveMaxTimeDuration() { - ServiceBusReceiverAsyncClient receiver = new ServiceBusClientBuilder() - .connectionString("fake-string") - .receiver() - .queueName("<< QUEUE NAME >>") - .buildAsyncClient(); - - // BEGIN: com.azure.messaging.servicebus.servicebusasyncreceiverclient.receive#int-duration - Disposable subscription = receiver.receiveMessages(15, Duration.ofSeconds(30)) - .flatMap(context -> { - ServiceBusReceivedMessage message = context.getMessage(); - System.out.printf("Received message id: %s%n", message.getMessageId()); - System.out.printf("Contents of message as string: %s%n", new String(message.getBody(), UTF_8)); - return receiver.complete(message.getLockToken()); - }).subscribe(aVoid -> System.out.println("Processed message."), - error -> System.out.println("Error occurred: " + error)); - - // subscribe is a non-blocking call and program flow will continue before messages are fetched. - // END: com.azure.messaging.servicebus.servicebusasyncreceiverclient.receive#int-duration - - // When program ends, or you're done receiving all messages. - subscription.dispose(); - - receiver.close(); - } - /** * Demonstrates how to create a session receiver for a single, first available session. */ diff --git a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClientIntegrationTest.java b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClientIntegrationTest.java index 9a5fcf5c67ff..823605ce3297 100644 --- a/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClientIntegrationTest.java +++ b/sdk/servicebus/azure-messaging-servicebus/src/test/java/com/azure/messaging/servicebus/ServiceBusReceiverAsyncClientIntegrationTest.java @@ -933,44 +933,6 @@ void setAndGetSessionState(MessagingEntityType entityType) { messagesPending.decrementAndGet(); } - @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityProvider") - @ParameterizedTest - void receivesByNumber(MessagingEntityType entityType) { - // Arrange - setSenderAndReceiver(entityType, TestUtils.USE_CASE_RECEIVE_BY_NUMBER, false); - - final String messageId = UUID.randomUUID().toString(); - final int number = 10; - final List messages = TestUtils.getServiceBusMessages(number, messageId, CONTENTS_BYTES); - - sendMessage(messages).block(Duration.ofSeconds(10)); - - // Act & Assert - StepVerifier.create(receiveAndDeleteReceiver.receiveMessages(messages.size(), Duration.ofSeconds(15)) - .doOnNext(next -> messagesPending.decrementAndGet())) - .expectNextCount(number) - .verifyComplete(); - } - - @MethodSource("com.azure.messaging.servicebus.IntegrationTestBase#messagingEntityProvider") - @ParameterizedTest - void receivesByTime(MessagingEntityType entityType) { - // Arrange - - setSenderAndReceiver(entityType, TestUtils.USE_CASE_RECEIVE_BY_TIME, false); - final String messageId = UUID.randomUUID().toString(); - final int number = 10; - final List messages = TestUtils.getServiceBusMessages(number, messageId, CONTENTS_BYTES); - - sendMessage(messages).block(Duration.ofSeconds(15)); - - // Act & Assert - StepVerifier.create(receiveAndDeleteReceiver.receiveMessages(number + 10, Duration.ofSeconds(15)) - .doOnNext(next -> messagesPending.decrementAndGet())) - .expectNextCount(number) - .verifyComplete(); - } - /** * Verifies that we can receive a message from dead letter queue. */