[Service Bus] Use processor in the readme (Azure#17386)

The processor client is what we should share in the getting started scenario and not the sync/async receiver clients. In this PR I have,

- used the processor client to receive messages instead of sync/async receiver clients
- removed the part on renewing locks as this is now done by default. These are remnants from the time when people had to opt in explicitly
- merged settling messages into the processor sample
- removed the session receiver example, put in just a note that it is similar to the non sessions case and listed out differences.

@srnagar I put together this PR in a haste, please make corrections as you see fit. I'd like to see the basic version checked in ASAP and you can make improvements in a different PR

Author: ramya-rao-a

sdk/servicebus/azure-messaging-servicebus/README.md

@@ -169,96 +169,44 @@ sender.sendMessages(messages);
 sender.close();
 ```
 
-### Receive messages and renew lock
+### Receive messages
 
-You'll need to create an asynchronous [`ServiceBusReceiverAsyncClient`][ServiceBusReceiverAsyncClient] or a synchronous
-[`ServiceBusReceiverClient`][ServiceBusReceiverClient] to receive messages. Each receiver can consume messages from
-either a queue or a topic subscription.
+To receive messages, you will need to create a `ServiceBusProcessorClient` with callbacks for incoming messages and any error that occurs in the process. You can then start and stop the client as required.
 
-By default, the receive mode is [`ReceiveMode.PEEK_LOCK`][ReceiveMode]. This tells the broker that the receiving client
-wants to manage settlement of received messages explicitly. The message is made available for the receiver to process,
-while held under an exclusive lock in the service so that other, competing receivers cannot see it.
-`ServiceBusReceivedMessage.getLockedUntil()` indicates when the lock expires and can be extended by the client using
-`receiver.renewMessageLock()`. A session lock can be extended by `receiver.renewSessionLock()` for a session enabled 
-queue or topic/subscriber.
+By default, the `autoComplete` feature is enabled on the processor client which means that after executing your callback for the message, the client will complete the message i.e. remove it from the queue/subscription. If your callback throws an error, then the client will abandon the message i.e. make it available to be received again. You can disable this feature when creating the processor client.
 
-#### Receive a batch of messages
-
-The snippet below creates a [`ServiceBusReceiverClient`][ServiceBusReceiverClient] to receive messages from a topic
-subscription.
-
-<!-- embedme ./src/samples/java/com/azure/messaging/servicebus/ReadmeSamples.java#L84-L101 -->
-```java
-ServiceBusReceiverClient receiver = new ServiceBusClientBuilder()
-    .connectionString("<< CONNECTION STRING FOR THE SERVICE BUS NAMESPACE >>")
-    .receiver()
-    .topicName("<< TOPIC NAME >>")
-    .subscriptionName("<< SUBSCRIPTION NAME >>")
-    .buildClient();
-
-// Receives a batch of messages when 10 messages are received or until 30 seconds have elapsed, whichever
-// happens first.
-IterableStream<ServiceBusReceivedMessage> messages = receiver.receiveMessages(10, Duration.ofSeconds(30));
-messages.forEach(message -> {
-
-    System.out.printf("Id: %s. Contents: %s%n", message.getMessageId(),
-        message.getBody().toString());
-});
-
-// When you are done using the receiver, dispose of it.
-receiver.close();
-```
-
-#### Receive a stream of messages
-
-The asynchronous [`ServiceBusReceiverAsyncClient`][ServiceBusReceiverAsyncClient] continuously fetches messages until
-the `subscription` is disposed.
-
-<!-- embedme ./src/samples/java/com/azure/messaging/servicebus/ReadmeSamples.java#L108-L130 -->
-```java
-ServiceBusReceiverAsyncClient receiver = new ServiceBusClientBuilder()
-    .connectionString("<< CONNECTION STRING FOR THE SERVICE BUS NAMESPACE >>")
-    .receiver()
-    .queueName("<< QUEUE NAME >>")
-    .buildAsyncClient();
-
-// receive() operation continuously fetches messages until the subscription is disposed.
-// The stream is infinite, and completes when the subscription or receiver is closed.
-Disposable subscription = receiver.receiveMessages().subscribe(message -> {
-
-    System.out.printf("Id: %s%n", message.getMessageId());
-    System.out.printf("Contents: %s%n", message.getBody().toString());
-}, error -> {
-        System.err.println("Error occurred while receiving messages: " + error);
-    }, () -> {
-        System.out.println("Finished receiving messages.");
-    });
-
-// Continue application processing. When you are finished receiving messages, dispose of the subscription.
-subscription.dispose();
-
-// When you are done using the receiver, dispose of it.
-receiver.close();
-```
-
-### Settle messages
-
-When a message is received, it can be settled using any of the `complete()`, `abandon()`, `defer()`, or `deadLetter()`
-overloads. The sample below completes a received message from synchronous
-[`ServiceBusReceiverClient`][ServiceBusReceiverClient].
-
-<!-- embedme ./src/samples/java/com/azure/messaging/servicebus/ReadmeSamples.java#L145-L151 -->
 ```java
-// This fetches a batch of 10 messages or until the default operation timeout has elapsed.
-receiver.receiveMessages(10).forEach(message -> {
-    // Process message and then complete it.
-    System.out.println("Completing message " + message.getLockToken());
-
-    receiver.complete(message);
-});
+// Sample code that processes a single message
+Consumer<ServiceBusReceivedMessageContext> processMessage = messageContext -> {
+    try {
+        System.out.println(messageContext.getMessage().getMessageId());
+        // other message processing code
+        messageContext.complete(); 
+    } catch (Exception ex) {
+        messageContext.abandon(); 
+    }
+}   
+
+// Sample code that gets called if there's an error
+Consumer<Throwable> processError = throwable -> {
+    logError(throwable);
+    metrics.recordError(throwable);
+}
+
+// create the processor client via the builder and its sub-builder
+ServiceBusProcessorClient processorClient = new ServiceBusClientBuilder()
+                                .connectionString("connection-string")
+                                .processor()
+                                .queueName("queue-name")
+                                .processMessage(processMessage)
+                                .processError(processError)
+                                .buildProcessorClient();
+
+// Starts the processor in the background and returns immediately
+processorClient.start();
 ```
 
-There are four ways of settling messages:
+There are four ways of settling messages using the methods on the message context passed to your callback.
   - Complete - causes the message to be deleted from the queue or topic.
   - Abandon - releases the receiver's lock on the message allowing for the message to be received by other receivers.
   - Defer - defers the message from being received by normal means. In order to receive deferred messages, the sequence
@@ -298,29 +246,9 @@ sender.sendMessage(message);
 
 #### Receive messages from a session
 
-Receivers can fetch messages from a specific session or the first available, unlocked session.
+Receiving messages from sessions is similar to receiving messages from a non session enabled queue or subscription. The difference is in the builder and the class you use.
 
-<!-- embedme ./src/samples/java/com/azure/messaging/servicebus/ReadmeSamples.java#L175-L181 -->
-```java
-// Creates a session-enabled receiver that gets messages from the session "greetings".
-ServiceBusSessionReceiverAsyncClient sessionReceiver = new ServiceBusClientBuilder()
-    .connectionString("<< CONNECTION STRING FOR THE SERVICE BUS NAMESPACE >>")
-    .sessionReceiver()
-    .queueName("<< QUEUE NAME >>")
-    .buildAsyncClient();
-Mono<ServiceBusReceiverAsyncClient> receiverAsyncClient = sessionReceiver.acceptSession("greetings");
-```
-
-<!-- embedme ./src/samples/java/com/azure/messaging/servicebus/ReadmeSamples.java#L188-L194 -->
-```java
-// Creates a session-enabled receiver that gets messages from the first available session.
-ServiceBusSessionReceiverAsyncClient sessionReceiver = new ServiceBusClientBuilder()
-    .connectionString("<< CONNECTION STRING FOR THE SERVICE BUS NAMESPACE >>")
-    .sessionReceiver()
-    .queueName("<< QUEUE NAME >>")
-    .buildAsyncClient();
-Mono<ServiceBusReceiverAsyncClient> receiverAsyncClient = sessionReceiver.acceptNextSession();
-```
+In non-session case, you would use the sub builder `processor()`. In case of sessions, you would use the sub builder `sessionProcessor()`. Both sub builders will create an instance of `ServiceBusProcessorClient` configured to work on a session or a non-session Service Bus entity. In the case of the session processor, you can pass the maximum number of sessions you want the processor to process concurrently as well.
 
 ### Create a dead-letter queue Receiver