-
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 support for settlement from the isolated worker extension (#38865)
* Add support for settlement from the isolated worker extension * Update comment * roll back version of Grpc.Tools * roll back further * PR fb * Revert eventArgs fields to private * remove in finally block * Fix test and move versions to package.data.props * Update sdk/servicebus/Microsoft.Azure.WebJobs.Extensions.ServiceBus/src/Listeners/ServiceBusListener.cs Co-authored-by: Jesse Squire <[email protected]> * Add batch test cases * Fix tests * Fix --------- Co-authored-by: Jesse Squire <[email protected]>
- Loading branch information
1 parent
f8ac921
commit 239c373
Showing
15 changed files
with
1,031 additions
and
54 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
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
28 changes: 28 additions & 0 deletions
28
...ebus/Microsoft.Azure.WebJobs.Extensions.ServiceBus/src/Grpc/Proto/SettlementExtensions.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,28 @@ | ||
| // Copyright (c) Microsoft Corporation. All rights reserved. | ||
| // Licensed under the MIT License. | ||
|
|
||
| #if NET6_0_OR_GREATER | ||
| using Microsoft.Azure.ServiceBus.Grpc; | ||
|
|
||
| namespace Microsoft.Azure.WebJobs.Extensions.ServiceBus.Grpc | ||
| { | ||
| internal static class SettlementExtensions | ||
| { | ||
| internal static object GetPropertyValue(this SettlementProperties properties) | ||
| { | ||
| return properties.ValuesCase switch | ||
| { | ||
| SettlementProperties.ValuesOneofCase.LongValue => properties.LongValue, | ||
| SettlementProperties.ValuesOneofCase.UlongValue => properties.UlongValue, | ||
| SettlementProperties.ValuesOneofCase.DoubleValue => properties.DoubleValue, | ||
| SettlementProperties.ValuesOneofCase.FloatValue => properties.FloatValue, | ||
| SettlementProperties.ValuesOneofCase.IntValue => properties.IntValue, | ||
| SettlementProperties.ValuesOneofCase.UintValue => properties.UintValue, | ||
| SettlementProperties.ValuesOneofCase.BoolValue => properties.BoolValue, | ||
| SettlementProperties.ValuesOneofCase.StringValue => properties.StringValue, | ||
| _ => null | ||
| }; | ||
| } | ||
| } | ||
| } | ||
| #endif |
66 changes: 66 additions & 0 deletions
66
sdk/servicebus/Microsoft.Azure.WebJobs.Extensions.ServiceBus/src/Grpc/Proto/settlement.proto
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,66 @@ | ||
| syntax = "proto3"; | ||
|
|
||
| import "google/protobuf/timestamp.proto"; | ||
| import "google/protobuf/empty.proto"; | ||
|
|
||
| // this namespace will be shared between isolated worker and WebJobs extension so make it somewhat generic | ||
| option csharp_namespace = "Microsoft.Azure.ServiceBus.Grpc"; | ||
|
|
||
| // The settlement service definition. | ||
| service Settlement { | ||
| // Completes a message | ||
| rpc Complete (CompleteRequest) returns (google.protobuf.Empty) {} | ||
|
|
||
| // Abandons a message | ||
| rpc Abandon (AbandonRequest) returns (google.protobuf.Empty) {} | ||
|
|
||
| // Deadletters a message | ||
| rpc Deadletter (DeadletterRequest) returns (google.protobuf.Empty) {} | ||
|
|
||
| // Defers a message | ||
| rpc Defer (DeferRequest) returns (google.protobuf.Empty) {} | ||
| } | ||
|
|
||
| // The complete message request containing the locktoken. | ||
| message CompleteRequest { | ||
| string locktoken = 1; | ||
| } | ||
|
|
||
| // The abandon message request containing the locktoken and properties to modify. | ||
| message AbandonRequest { | ||
| string locktoken = 1; | ||
| map<string, SettlementProperties> propertiesToModify = 2; | ||
| } | ||
|
|
||
| // The deadletter message request containing the locktoken and properties to modify along with the reason/description. | ||
| message DeadletterRequest { | ||
| string locktoken = 1; | ||
| map<string, SettlementProperties> propertiesToModify = 2; | ||
| string deadletterReason = 3; | ||
| string deadletterErrorDescription = 4; | ||
| } | ||
|
|
||
| // The defer message request containing the locktoken and properties to modify. | ||
| message DeferRequest { | ||
| string locktoken = 1; | ||
| map<string, SettlementProperties> propertiesToModify = 2; | ||
| } | ||
|
|
||
| // The settlement property can be of any type listed below, which | ||
| // corresponds to the types specified in | ||
| // https://learn.microsoft.com/en-us/dotnet/api/azure.messaging.servicebus.servicebusmessage.applicationproperties?view=azure-dotnet#remarks | ||
| // Note: this list doesn't match 1:1 with the supported Service Bus types, so compatible types are used in some cases - e.g. | ||
| // short uses int, TimeSpan uses string, etc. The full list of transforms can be found in the isolated worker extension source code. | ||
| message SettlementProperties { | ||
| oneof values { | ||
| string stringValue = 1; | ||
| int32 intValue = 2; | ||
| uint32 uintValue = 3; | ||
| int64 longValue = 4; | ||
| uint64 ulongValue = 5; | ||
| bool boolValue = 6; | ||
| float floatValue = 7; | ||
| double doubleValue = 8; | ||
| google.protobuf.Timestamp timestampValue = 9; | ||
| } | ||
| } |
89 changes: 89 additions & 0 deletions
89
sdk/servicebus/Microsoft.Azure.WebJobs.Extensions.ServiceBus/src/Grpc/SettlementService.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,89 @@ | ||
| // Copyright (c) Microsoft Corporation. All rights reserved. | ||
| // Licensed under the MIT License. | ||
|
|
||
| #if NET6_0_OR_GREATER | ||
| using System; | ||
| using System.Linq; | ||
| using System.Threading.Tasks; | ||
| using Google.Protobuf.WellKnownTypes; | ||
| using Grpc.Core; | ||
| using Microsoft.Azure.ServiceBus.Grpc; | ||
| using Microsoft.Azure.WebJobs.ServiceBus; | ||
|
|
||
| namespace Microsoft.Azure.WebJobs.Extensions.ServiceBus.Grpc | ||
| { | ||
| internal class SettlementService : Settlement.SettlementBase | ||
| { | ||
| private readonly MessagingProvider _provider; | ||
|
|
||
| public SettlementService(MessagingProvider provider) | ||
| { | ||
| _provider = provider; | ||
| } | ||
|
|
||
| public SettlementService() | ||
| { | ||
| _provider = null; | ||
| } | ||
|
|
||
| public override async Task<Empty> Complete(CompleteRequest request, ServerCallContext context) | ||
| { | ||
| if (_provider.ActionsCache.TryGetValue(request.Locktoken, out var tuple)) | ||
| { | ||
| await tuple.Actions.CompleteMessageAsync( | ||
| tuple.Message, | ||
| context.CancellationToken).ConfigureAwait(false); | ||
| return new Empty(); | ||
| } | ||
| throw new RpcException (new Status(StatusCode.FailedPrecondition, $"LockToken {request.Locktoken} not found.")); | ||
| } | ||
|
|
||
| public override async Task<Empty> Abandon(AbandonRequest request, ServerCallContext context) | ||
| { | ||
| if (_provider.ActionsCache.TryGetValue(request.Locktoken, out var tuple)) | ||
| { | ||
| await tuple.Actions.AbandonMessageAsync( | ||
| tuple.Message, | ||
| request.PropertiesToModify.ToDictionary( | ||
| pair => pair.Key, | ||
| pair => pair.Value.GetPropertyValue()), | ||
| context.CancellationToken).ConfigureAwait(false); | ||
| return new Empty(); | ||
| } | ||
| throw new RpcException (new Status(StatusCode.FailedPrecondition, $"LockToken {request.Locktoken} not found.")); | ||
| } | ||
|
|
||
| public override async Task<Empty> Defer(DeferRequest request, ServerCallContext context) | ||
| { | ||
| if (_provider.ActionsCache.TryGetValue(request.Locktoken, out var tuple)) | ||
| { | ||
| await tuple.Actions.DeferMessageAsync( | ||
| tuple.Message, | ||
| request.PropertiesToModify.ToDictionary( | ||
| pair => pair.Key, | ||
| pair => pair.Value.GetPropertyValue()), | ||
| context.CancellationToken).ConfigureAwait(false); | ||
| return new Empty(); | ||
| } | ||
| throw new RpcException (new Status(StatusCode.FailedPrecondition, $"LockToken {request.Locktoken} not found.")); | ||
| } | ||
|
|
||
| public override async Task<Empty> Deadletter(DeadletterRequest request, ServerCallContext context) | ||
| { | ||
| if (_provider.ActionsCache.TryGetValue(request.Locktoken, out var tuple)) | ||
| { | ||
| await tuple.Actions.DeadLetterMessageAsync( | ||
| tuple.Message, | ||
| request.PropertiesToModify.ToDictionary( | ||
| pair => pair.Key, | ||
| pair => pair.Value.GetPropertyValue()), | ||
| request.DeadletterReason, | ||
| request.DeadletterErrorDescription, | ||
| context.CancellationToken).ConfigureAwait(false); | ||
| return new Empty(); | ||
| } | ||
| throw new RpcException (new Status(StatusCode.FailedPrecondition, $"LockToken {request.Locktoken} not found.")); | ||
| } | ||
| } | ||
| } | ||
| #endif |
Oops, something went wrong.