-
Notifications
You must be signed in to change notification settings - Fork 4.8k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* Add samples demonstrating cross receiver settlement * Fix * Fix * fix line break * Fix sln * regenerate * Apply suggestions from code review Co-authored-by: Jesse Squire <[email protected]> --------- Co-authored-by: Jesse Squire <[email protected]>
- Loading branch information
1 parent
f37809b
commit ad2201a
Showing
4 changed files
with
162 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
67 changes: 67 additions & 0 deletions
67
...s/Azure.Messaging.ServiceBus/samples/Sample16_CrossReceiverMessageSettlement.md
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,67 @@ | ||
| # Cross Receiver Message Settlement | ||
|
|
||
| The message settlement APIs on the `ServiceBusReceiver` require passing in the | ||
| `ServiceBusReceivedMessage`. This can be limiting for scenarios in which the message needs to be persisted and then | ||
| settled across process boundaries. There are two strategies that can be used for settling a message with a different | ||
| receiver. The recommended strategy depends on the specific application scenario. | ||
|
|
||
| ## Storing the entire `ServiceBusReceivedMessage` | ||
|
|
||
| If it is necessary to store the entire message and then rehydrate it in another process, use the following strategy. | ||
| First we can get the raw AMQP message bytes and lock token as shown below: | ||
| *Note: The lock token is not | ||
| actually part of the AMQP message so it needs to stored separately from the AMQP bytes.* | ||
|
|
||
| ```C# Snippet:ServiceBusWriteReceivedMessage | ||
| var client1 = new ServiceBusClient(connectionString); | ||
| ServiceBusSender sender = client1.CreateSender(queueName); | ||
|
|
||
| var message = new ServiceBusMessage("some message"); | ||
| await sender.SendMessageAsync(message); | ||
|
|
||
| ServiceBusReceiver receiver1 = client1.CreateReceiver(queueName); | ||
| ServiceBusReceivedMessage receivedMessage = await receiver1.ReceiveMessageAsync(); | ||
| ReadOnlyMemory<byte> amqpMessageBytes = receivedMessage.GetRawAmqpMessage().ToBytes().ToMemory(); | ||
| ReadOnlyMemory<byte> lockTokenBytes = Guid.Parse(receivedMessage.LockToken).ToByteArray(); | ||
| ``` | ||
|
|
||
| In order to rehydrate the message in another process, we would do the following: | ||
|
|
||
| ```C# Snippet:ServiceBusReadReceivedMessage | ||
| AmqpAnnotatedMessage amqpMessage = AmqpAnnotatedMessage.FromBytes(new BinaryData(amqpMessageBytes)); | ||
| ServiceBusReceivedMessage rehydratedMessage = ServiceBusReceivedMessage.FromAmqpMessage(amqpMessage, new BinaryData(lockTokenBytes)); | ||
|
|
||
| var client2 = new ServiceBusClient(connectionString); | ||
| ServiceBusReceiver receiver2 = client2.CreateReceiver(queueName); | ||
| await receiver2.CompleteMessageAsync(rehydratedMessage); | ||
| ``` | ||
|
|
||
| ## Storing only the lock token | ||
|
|
||
| If the entire message is not needed when settling the message in a different process, you can simply preserve the | ||
| lock token. In the example below, we store off the lock token using its GUID bytes. You can also simply store a | ||
| string if that is easier for your scenario. | ||
|
|
||
| ```C# Snippet:ServiceBusWriteReceivedMessageLockToken | ||
| var client1 = new ServiceBusClient(connectionString); | ||
| ServiceBusSender sender = client1.CreateSender(queueName); | ||
|
|
||
| var message = new ServiceBusMessage("some message"); | ||
| await sender.SendMessageAsync(message); | ||
|
|
||
| ServiceBusReceiver receiver1 = client1.CreateReceiver(queueName); | ||
| ServiceBusReceivedMessage receivedMessage = await receiver1.ReceiveMessageAsync(); | ||
| ReadOnlyMemory<byte> lockTokenBytes = Guid.Parse(receivedMessage.LockToken).ToByteArray(); | ||
| ``` | ||
|
|
||
| In order to rehydrate the message in another process using the lock token, we would do the following: | ||
| *Note: Because we only stored the lock token, when we rehydrate the message all of the properties of the | ||
| `ServiceBusReceivedMessage` other than `LockToken` will have default values.* | ||
|
|
||
| ```C# Snippet:ServiceBusReadReceivedMessageLockToken | ||
| ServiceBusReceivedMessage rehydratedMessage = ServiceBusModelFactory.ServiceBusReceivedMessage(lockTokenGuid: new Guid(lockTokenBytes.ToArray())); | ||
|
|
||
| var client2 = new ServiceBusClient(connectionString); | ||
| ServiceBusReceiver receiver2 = client2.CreateReceiver(queueName); | ||
| await receiver2.CompleteMessageAsync(rehydratedMessage); | ||
| ``` |
91 changes: 91 additions & 0 deletions
91
...cebus/Azure.Messaging.ServiceBus/tests/Samples/Sample16_CrossReceiverMessageSettlement.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,91 @@ | ||
| // Copyright (c) Microsoft Corporation. All rights reserved. | ||
| // Licensed under the MIT License. | ||
|
|
||
| using System; | ||
| using System.Collections.Generic; | ||
| using System.Threading.Tasks; | ||
| using Azure.Core.Amqp; | ||
| using NUnit.Framework; | ||
|
|
||
| namespace Azure.Messaging.ServiceBus.Tests.Samples | ||
| { | ||
| public class Sample16_CrossReceiverMessageSettlement : ServiceBusLiveTestBase | ||
| { | ||
| [Test] | ||
| public async Task RehydrateReceivedMessageUsingRawBytes() | ||
| { | ||
| await using (var scope = await ServiceBusScope.CreateWithQueue(enablePartitioning: false, enableSession: false)) | ||
| { | ||
| #if SNIPPET | ||
| string connectionString = "<connection_string>"; | ||
| string queueName = "<queue_name>"; | ||
| #else | ||
| string connectionString = TestEnvironment.ServiceBusConnectionString; | ||
| string queueName = scope.QueueName; | ||
| #endif | ||
|
|
||
| #region Snippet:ServiceBusWriteReceivedMessage | ||
|
|
||
| var client1 = new ServiceBusClient(connectionString); | ||
| ServiceBusSender sender = client1.CreateSender(queueName); | ||
|
|
||
| var message = new ServiceBusMessage("some message"); | ||
| await sender.SendMessageAsync(message); | ||
|
|
||
| ServiceBusReceiver receiver1 = client1.CreateReceiver(queueName); | ||
| ServiceBusReceivedMessage receivedMessage = await receiver1.ReceiveMessageAsync(); | ||
| ReadOnlyMemory<byte> amqpMessageBytes = receivedMessage.GetRawAmqpMessage().ToBytes().ToMemory(); | ||
| ReadOnlyMemory<byte> lockTokenBytes = Guid.Parse(receivedMessage.LockToken).ToByteArray(); | ||
| #endregion | ||
|
|
||
| #region Snippet:ServiceBusReadReceivedMessage | ||
| AmqpAnnotatedMessage amqpMessage = AmqpAnnotatedMessage.FromBytes(new BinaryData(amqpMessageBytes)); | ||
| ServiceBusReceivedMessage rehydratedMessage = ServiceBusReceivedMessage.FromAmqpMessage(amqpMessage, new BinaryData(lockTokenBytes)); | ||
|
|
||
| var client2 = new ServiceBusClient(connectionString); | ||
| ServiceBusReceiver receiver2 = client2.CreateReceiver(queueName); | ||
| await receiver2.CompleteMessageAsync(rehydratedMessage); | ||
| #endregion | ||
|
|
||
| Assert.AreEqual("some message", rehydratedMessage.Body.ToString()); | ||
| } | ||
| } | ||
|
|
||
| [Test] | ||
| public async Task RehydrateReceivedMessageUsingLockToken() | ||
| { | ||
| await using (var scope = await ServiceBusScope.CreateWithQueue(enablePartitioning: false, enableSession: false)) | ||
| { | ||
| #if SNIPPET | ||
| string connectionString = "<connection_string>"; | ||
| string queueName = "<queue_name>"; | ||
| #else | ||
| string connectionString = TestEnvironment.ServiceBusConnectionString; | ||
| string queueName = scope.QueueName; | ||
| #endif | ||
|
|
||
| #region Snippet:ServiceBusWriteReceivedMessageLockToken | ||
|
|
||
| var client1 = new ServiceBusClient(connectionString); | ||
| ServiceBusSender sender = client1.CreateSender(queueName); | ||
|
|
||
| var message = new ServiceBusMessage("some message"); | ||
| await sender.SendMessageAsync(message); | ||
|
|
||
| ServiceBusReceiver receiver1 = client1.CreateReceiver(queueName); | ||
| ServiceBusReceivedMessage receivedMessage = await receiver1.ReceiveMessageAsync(); | ||
| ReadOnlyMemory<byte> lockTokenBytes = Guid.Parse(receivedMessage.LockToken).ToByteArray(); | ||
| #endregion | ||
|
|
||
| #region Snippet:ServiceBusReadReceivedMessageLockToken | ||
|
|
||
| ServiceBusReceivedMessage rehydratedMessage = ServiceBusModelFactory.ServiceBusReceivedMessage(lockTokenGuid: new Guid(lockTokenBytes.ToArray())); | ||
|
|
||
| var client2 = new ServiceBusClient(connectionString); | ||
| ServiceBusReceiver receiver2 = client2.CreateReceiver(queueName); | ||
| await receiver2.CompleteMessageAsync(rehydratedMessage); | ||
| #endregion | ||
| } | ||
| } | ||
| } | ||
| } |