+ * This method continues to receive messages from the dead letter queue, then completes them.
+ *
+ *
+ * @param builder Service Bus client builder.
+ * @param queueName Name of the queue to receive from.
+ *
+ * @return A Mono that completes when all messages in queue have been processed (because a message has not been
+ * received in the last 30 seconds).
+ */
+ private Mono receiveAndCompleteDeadLetterQueueMessages(ServiceBusClientBuilder builder, String queueName) {
+ return Mono.using(() -> {
+ return builder.receiver()
+ .queueName(queueName)
+ .subQueue(SubQueue.DEAD_LETTER_QUEUE)
+ .buildAsyncClient();
+ }, deadLetterQueueReceiver -> {
+ // Continue to receive messages until no message has been seen for 30 seconds.
+ return deadLetterQueueReceiver.receiveMessages()
+ .timeout(Duration.ofSeconds(30))
+ .flatMap(message -> {
+ System.out.printf("Received message from dead-letter queue. Sequence #: %s. DeliveryCount %d%n",
+ message.getSequenceNumber(), message.getDeliveryCount());
+
+ // When messages are dead lettered, there are system properties that can be set to denote the
+ // reason why via fields such as dead-letter reason and dead-letter description.
+ System.out.printf("Dead-Letter Reason: %s. Description: %s. Source: %s%n",
+ message.getDeadLetterReason(), message.getDeadLetterErrorDescription(),
+ message.getDeadLetterSource());
+
+ System.out.println("Application properties:");
+ message.getApplicationProperties().forEach((key, value) ->
+ System.out.printf("\t%s=%s%n", key, value));
+
+ // We complete the messages we received from the queue.
+ // In a real example, you may do some sort of computation to understand why previous attempts to
+ // process the message failed. This is shown in Scenario 2.
+ return deadLetterQueueReceiver.complete(message);
+ })
+ .onErrorResume(TimeoutException.class, exception -> {
+ System.out.println("No messages received after 30 seconds. Dead-letter queue is empty.");
+ return Mono.empty();
+ })
+ .then();
+ }, deadLetterQueueReceiver -> {
+ // When the receiving operation is completed, close the client.
+ deadLetterQueueReceiver.close();
+ });
+ }
+
+ /**
+ * Scenario 2: Part 1
+ *
+ *
+ * Receives {@link ServiceBusMessage messages} and dead letters them if it has a subject of "Scientist" and content
+ * type of "application/json". This is to simulate that the message may be malformed or didn't contain the right
+ * content, so we dead letter it so other receivers don't process this message.
+ *
+ *
+ * @param builder Service Bus client builder.
+ * @param queueName Name of queue to receive messages from.
+ *
+ * @return A Mono that completes when all the messages in the queue have been processed (because a message has not
+ * been received in the last 30 seconds).
+ */
+ private Mono receiveAndDeadletterMessages(ServiceBusClientBuilder builder, String queueName) {
+ return Mono.using(
+ () -> {
+ // Creates the async receiver.
+ return builder.receiver()
+ .queueName(queueName)
+ .receiveMode(ServiceBusReceiveMode.PEEK_LOCK)
+ .buildAsyncClient();
+ },
+ receiver -> {
+ // The scope of the receiver resource is to receive messages for a duration of `receiveDuration` then
+ // complete or dead-letter the message based on its content-type and subject.
+ return receiver.receiveMessages()
+ .timeout(Duration.ofSeconds(30))
+ .flatMap(message -> {
+ final String subject = message.getSubject();
+ final String contentType = message.getContentType();
+ final Person person;
+ try {
+ person = Person.fromJson(message.getBody().toString());
+ } catch (RuntimeException e) {
+ return Mono.error(new RuntimeException("Could not deserialize message: "
+ + message.getSequenceNumber(), e));
+ }
+
+ System.out.printf("Received message. SequenceNumber = %s. EnqueuedTimeUtc = %s. "
+ + "ExpiresAtUtc = %s. ContentType = %s. Content: [ %s ]%n",
+ message.getSequenceNumber(), message.getEnqueuedTime(), message.getExpiresAt(),
+ message.getContentType(), person);
+
+ // Simulates that something could not be processed in the message, so we dead letter it.
+ if ("Scientist".equals(subject) && "application/json".equals(contentType)) {
+ return receiver.complete(message);
+ } else {
+ return receiver.deadLetter(message);
+ }
+ })
+ .onErrorResume(TimeoutException.class, exception -> {
+ System.out.println("No messages received after 30 seconds. Queue is empty.");
+ return Mono.empty();
+ })
+ .then();
+ },
+ receiver -> {
+ // When the receiving operation is completed, close the client.
+ receiver.close();
+ });
+ }
+
+ /**
+ * Scenario 2: Part 2
+ *
+ *
+ * Receives {@link ServiceBusMessage messages} from the dead letter queue, it fixes up the message then resends the
+ * fixed message to the queue again. This simulates messages that may have errors in them, were dead-lettered,
+ * and reprocessed in the dead-letter queue so the data is correct again.
+ *
+ *
+ * @param builder Service Bus client builder.
+ * @param queueName Name of queue to receive messages from.
+ * @param resubmitSender Service Bus sender client. When messages are fixed, they are published via this sender.
+ *
+ * @return A Mono that completes when all the messages in the queue have been processed (because a message has not
+ * been received in the last 30 seconds).
+ */
+ private Mono receiveAndFixDeadLetterQueueMessages(ServiceBusClientBuilder builder, String queueName,
+ ServiceBusSenderClient resubmitSender) {
+
+ return Mono.using(() -> {
+ return builder.receiver()
+ .queueName(queueName)
+ .subQueue(SubQueue.DEAD_LETTER_QUEUE)
+ .buildAsyncClient();
+ }, deadLetterQueueReceiver -> {
+ // Continue to receive messages until no message has been seen for 30 seconds.
+ return deadLetterQueueReceiver.receiveMessages()
+ .timeout(Duration.ofSeconds(30))
+ .flatMap(message -> {
+ if ("Physicist".equals(message.getSubject())) {
+ System.out.printf("Fixing DLQ message. MessageId = %s. SequenceNumber = %s. Subject = %s%n",
+ message.getMessageId(), message.getSequenceNumber(), message.getSubject());
+
+ // Create a copy of this message and then set the subject to the correct one.
+ ServiceBusMessage resubmitMessage = new ServiceBusMessage(message)
+ .setSubject("Scientist");
+
+ // Sending that corrected message.
+ resubmitSender.sendMessage(resubmitMessage);
+ } else {
+ System.out.printf("Message resubmission is not required. MessageId = %s. SequenceNumber = %s. "
+ + "Subject = %s%n",
+ message.getMessageId(), message.getSequenceNumber(), message.getSubject());
+ }
+
+ return deadLetterQueueReceiver.complete(message);
+ })
+ .onErrorResume(TimeoutException.class, exception -> {
+ System.out.println("No messages received after 30 seconds. Dead-letter queue is empty.");
+ return Mono.empty();
+ })
+ .then();
+ }, deadLetterQueueReceiver -> {
+ // When the receiving operation is completed, close the client.
+ deadLetterQueueReceiver.close();
+ });
+ }
+
+ private static final class Person {
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ private final String lastName;
+ private final String firstName;
+
+ Person(@JsonProperty String lastName, @JsonProperty String firstName) {
+ this.lastName = lastName;
+ this.firstName = firstName;
+ }
+
+ String getLastName() {
+ return lastName;
+ }
+
+ String getFirstName() {
+ return firstName;
+ }
+
+ /**
+ * Serializes an item into its JSON string equivalent.
+ *
+ * @return The JSON representation.
+ *
+ * @throws RuntimeException if the person could not be serialized.
+ */
+ String toJson() {
+ try {
+ return OBJECT_MAPPER.writeValueAsString(this);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException("Could not serialize object.", e);
+ }
+ }
+
+ /**
+ * Deserializes a JSON string into a Person.
+ *
+ * @return The corresponding person.
+ *
+ * @throws RuntimeException if the JSON string could not be deserialized.
+ */
+ private static Person fromJson(String json) {
+ try {
+ return OBJECT_MAPPER.readValue(json, Person.class);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException("Could not deserialize object.", e);
+ }
+ }
+ }
+}