-
Notifications
You must be signed in to change notification settings - Fork 5.8k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Dotnet Grpc worker implementation (#5245)
Co-authored-by: Jacob Alber <[email protected]> Co-authored-by: Ryan Sweet <[email protected]>
- Loading branch information
3 people
authored
Feb 5, 2025
1 parent
9030f75
commit 08f9830
Showing
35 changed files
with
1,861 additions
and
269 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,34 @@ | ||
| // Copyright (c) Microsoft Corporation. All rights reserved. | ||
| // Checker.cs | ||
|
|
||
| using Microsoft.AutoGen.Contracts; | ||
| using Microsoft.AutoGen.Core; | ||
| using Microsoft.Extensions.Hosting; | ||
| using TerminationF = System.Func<int, bool>; | ||
|
|
||
| namespace GettingStartedGrpcSample; | ||
|
|
||
| [TypeSubscription("default")] | ||
| public class Checker( | ||
| AgentId id, | ||
| IAgentRuntime runtime, | ||
| IHostApplicationLifetime hostApplicationLifetime, | ||
| TerminationF runUntilFunc | ||
| ) : | ||
| BaseAgent(id, runtime, "Modifier", null), | ||
| IHandle<Events.CountUpdate> | ||
| { | ||
| public async ValueTask HandleAsync(Events.CountUpdate item, MessageContext messageContext) | ||
| { | ||
| if (!runUntilFunc(item.NewCount)) | ||
| { | ||
| Console.WriteLine($"\nChecker:\n{item.NewCount} passed the check, continue."); | ||
| await this.PublishMessageAsync(new Events.CountMessage { Content = item.NewCount }, new TopicId("default")); | ||
| } | ||
| else | ||
| { | ||
| Console.WriteLine($"\nChecker:\n{item.NewCount} failed the check, stopping."); | ||
| hostApplicationLifetime.StopApplication(); | ||
| } | ||
| } | ||
| } |
26 changes: 26 additions & 0 deletions
26
dotnet/samples/GettingStartedGrpc/GettingStartedGrpc.csproj
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,26 @@ | ||
| <Project Sdk="Microsoft.NET.Sdk"> | ||
|
|
||
| <PropertyGroup> | ||
| <OutputType>Exe</OutputType> | ||
| <TargetFramework>net8.0</TargetFramework> | ||
| <RootNamespace>getting_started</RootNamespace> | ||
| <ImplicitUsings>enable</ImplicitUsings> | ||
| <Nullable>enable</Nullable> | ||
| </PropertyGroup> | ||
|
|
||
| <ItemGroup> | ||
| <ProjectReference Include="..\..\src\Microsoft.AutoGen\Contracts\Microsoft.AutoGen.Contracts.csproj" /> | ||
| <ProjectReference Include="..\..\src\Microsoft.AutoGen\Core\Microsoft.AutoGen.Core.csproj" /> | ||
| <ProjectReference Include="..\..\src\Microsoft.AutoGen\Core.Grpc\Microsoft.AutoGen.Core.Grpc.csproj" /> | ||
| </ItemGroup> | ||
|
|
||
|
|
||
| <ItemGroup> | ||
| <Protobuf Include="message.proto" GrpcServices="Client" Link="Protos\message.proto" /> | ||
| </ItemGroup> | ||
|
|
||
| <ItemGroup> | ||
| <PackageReference Include="Grpc.Tools" PrivateAssets="All" /> | ||
| </ItemGroup> | ||
|
|
||
| </Project> |
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,29 @@ | ||
| // Copyright (c) Microsoft Corporation. All rights reserved. | ||
| // Modifier.cs | ||
|
|
||
| using Microsoft.AutoGen.Contracts; | ||
| using Microsoft.AutoGen.Core; | ||
|
|
||
| using ModifyF = System.Func<int, int>; | ||
|
|
||
| namespace GettingStartedGrpcSample; | ||
|
|
||
| [TypeSubscription("default")] | ||
| public class Modifier( | ||
| AgentId id, | ||
| IAgentRuntime runtime, | ||
| ModifyF modifyFunc | ||
| ) : | ||
| BaseAgent(id, runtime, "Modifier", null), | ||
| IHandle<Events.CountMessage> | ||
| { | ||
|
|
||
| public async ValueTask HandleAsync(Events.CountMessage item, MessageContext messageContext) | ||
| { | ||
| int newValue = modifyFunc(item.Content); | ||
| Console.WriteLine($"\nModifier:\nModified {item.Content} to {newValue}"); | ||
|
|
||
| var updateMessage = new Events.CountUpdate { NewCount = newValue }; | ||
| await this.PublishMessageAsync(updateMessage, topic: new TopicId("default")); | ||
| } | ||
| } |
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,36 @@ | ||
| // Copyright (c) Microsoft Corporation. All rights reserved. | ||
| // Program.cs | ||
| using GettingStartedGrpcSample; | ||
| using Microsoft.AutoGen.Contracts; | ||
| using Microsoft.AutoGen.Core; | ||
| using Microsoft.AutoGen.Core.Grpc; | ||
| using Microsoft.Extensions.DependencyInjection.Extensions; | ||
| using ModifyF = System.Func<int, int>; | ||
| using TerminationF = System.Func<int, bool>; | ||
|
|
||
| ModifyF modifyFunc = (int x) => x - 1; | ||
| TerminationF runUntilFunc = (int x) => | ||
| { | ||
| return x <= 1; | ||
| }; | ||
|
|
||
| AgentsAppBuilder appBuilder = new AgentsAppBuilder(); | ||
| appBuilder.AddGrpcAgentWorker("http://localhost:50051"); | ||
|
|
||
| appBuilder.Services.TryAddSingleton(modifyFunc); | ||
| appBuilder.Services.TryAddSingleton(runUntilFunc); | ||
|
|
||
| appBuilder.AddAgent<Checker>("Checker"); | ||
| appBuilder.AddAgent<Modifier>("Modifier"); | ||
|
|
||
| var app = await appBuilder.BuildAsync(); | ||
| await app.StartAsync(); | ||
|
|
||
| // Send the initial count to the agents app, running on the `local` runtime, and pass through the registered services via the application `builder` | ||
| await app.PublishMessageAsync(new GettingStartedGrpcSample.Events.CountMessage | ||
| { | ||
| Content = 10 | ||
| }, new TopicId("default")); | ||
|
|
||
| // Run until application shutdown | ||
| await app.WaitForShutdownAsync(); |
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,11 @@ | ||
| syntax = "proto3"; | ||
|
|
||
| option csharp_namespace = "GettingStartedGrpcSample.Events"; | ||
|
|
||
| message CountMessage { | ||
| int32 content = 1; | ||
| } | ||
|
|
||
| message CountUpdate { | ||
| int32 new_count = 1; | ||
| } |
76 changes: 76 additions & 0 deletions
76
dotnet/src/Microsoft.AutoGen/Core.Grpc/AgentsAppBuilderExtensions.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,76 @@ | ||
| // Copyright (c) Microsoft Corporation. All rights reserved. | ||
| // AgentsAppBuilderExtensions.cs | ||
|
|
||
| using System.Diagnostics; | ||
| using Grpc.Core; | ||
| using Grpc.Net.Client.Configuration; | ||
| using Microsoft.AutoGen.Contracts; | ||
| using Microsoft.AutoGen.Protobuf; | ||
| using Microsoft.Extensions.Configuration; | ||
| using Microsoft.Extensions.DependencyInjection; | ||
| using Microsoft.Extensions.DependencyInjection.Extensions; | ||
| using Microsoft.Extensions.Logging; | ||
| namespace Microsoft.AutoGen.Core.Grpc; | ||
|
|
||
| public static class AgentsAppBuilderExtensions | ||
| { | ||
| private const string _defaultAgentServiceAddress = "http://localhost:53071"; | ||
|
|
||
| // TODO: How do we ensure AddGrpcAgentWorker and UseInProcessRuntime are mutually exclusive? | ||
| public static AgentsAppBuilder AddGrpcAgentWorker(this AgentsAppBuilder builder, string? agentServiceAddress = null) | ||
| { | ||
| builder.Services.AddGrpcClient<AgentRpc.AgentRpcClient>(options => | ||
| { | ||
| options.Address = new Uri(agentServiceAddress ?? builder.Configuration.GetValue("AGENT_HOST", _defaultAgentServiceAddress)); | ||
| options.ChannelOptionsActions.Add(channelOptions => | ||
| { | ||
| var loggerFactory = new LoggerFactory(); | ||
| if (Debugger.IsAttached) | ||
| { | ||
| channelOptions.HttpHandler = new SocketsHttpHandler | ||
| { | ||
| EnableMultipleHttp2Connections = false, | ||
| KeepAlivePingDelay = TimeSpan.FromSeconds(200), | ||
| KeepAlivePingTimeout = TimeSpan.FromSeconds(100), | ||
| KeepAlivePingPolicy = HttpKeepAlivePingPolicy.Always | ||
| }; | ||
| } | ||
| else | ||
| { | ||
| channelOptions.HttpHandler = new SocketsHttpHandler | ||
| { | ||
| EnableMultipleHttp2Connections = true, | ||
| KeepAlivePingDelay = TimeSpan.FromSeconds(20), | ||
| KeepAlivePingTimeout = TimeSpan.FromSeconds(10), | ||
| KeepAlivePingPolicy = HttpKeepAlivePingPolicy.WithActiveRequests | ||
| }; | ||
| } | ||
|
|
||
| var methodConfig = new MethodConfig | ||
| { | ||
| Names = { MethodName.Default }, | ||
| RetryPolicy = new RetryPolicy | ||
| { | ||
| MaxAttempts = 5, | ||
| InitialBackoff = TimeSpan.FromSeconds(1), | ||
| MaxBackoff = TimeSpan.FromSeconds(5), | ||
| BackoffMultiplier = 1.5, | ||
| RetryableStatusCodes = { StatusCode.Unavailable } | ||
| } | ||
| }; | ||
|
|
||
| channelOptions.ServiceConfig = new() { MethodConfigs = { methodConfig } }; | ||
| channelOptions.ThrowOperationCanceledOnCancellation = true; | ||
| }); | ||
| }); | ||
|
|
||
| builder.Services.TryAddSingleton(DistributedContextPropagator.Current); | ||
| builder.Services.AddSingleton<IAgentRuntime, GrpcAgentRuntime>(); | ||
| builder.Services.AddHostedService<GrpcAgentRuntime>(services => | ||
| { | ||
| return (services.GetRequiredService<IAgentRuntime>() as GrpcAgentRuntime)!; | ||
| }); | ||
|
|
||
| return builder; | ||
| } | ||
| } |
43 changes: 43 additions & 0 deletions
43
dotnet/src/Microsoft.AutoGen/Core.Grpc/CloudEventExtensions.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,43 @@ | ||
| // Copyright (c) Microsoft Corporation. All rights reserved. | ||
| // CloudEventExtensions.cs | ||
|
|
||
| using Microsoft.AutoGen.Contracts; | ||
|
|
||
| namespace Microsoft.AutoGen.Core.Grpc; | ||
|
|
||
| internal static class CloudEventExtensions | ||
| { | ||
| // Convert an ISubscrptionDefinition to a Protobuf Subscription | ||
| internal static CloudEvent CreateCloudEvent(Google.Protobuf.WellKnownTypes.Any payload, TopicId topic, string dataType, AgentId? sender, string messageId) | ||
| { | ||
| var attributes = new Dictionary<string, CloudEvent.Types.CloudEventAttributeValue> | ||
| { | ||
| { | ||
| Constants.DATA_CONTENT_TYPE_ATTR, new CloudEvent.Types.CloudEventAttributeValue { CeString = Constants.DATA_CONTENT_TYPE_PROTOBUF_VALUE } | ||
| }, | ||
| { | ||
| Constants.DATA_SCHEMA_ATTR, new CloudEvent.Types.CloudEventAttributeValue { CeString = dataType } | ||
| }, | ||
| { | ||
| Constants.MESSAGE_KIND_ATTR, new CloudEvent.Types.CloudEventAttributeValue { CeString = Constants.MESSAGE_KIND_VALUE_PUBLISH } | ||
| } | ||
| }; | ||
|
|
||
| if (sender != null) | ||
| { | ||
| var senderNonNull = (AgentId)sender; | ||
| attributes.Add(Constants.AGENT_SENDER_TYPE_ATTR, new CloudEvent.Types.CloudEventAttributeValue { CeString = senderNonNull.Type }); | ||
| attributes.Add(Constants.AGENT_SENDER_KEY_ATTR, new CloudEvent.Types.CloudEventAttributeValue { CeString = senderNonNull.Key }); | ||
| } | ||
|
|
||
| return new CloudEvent | ||
| { | ||
| ProtoData = payload, | ||
| Type = topic.Type, | ||
| Source = topic.Source, | ||
| Id = messageId, | ||
| Attributes = { attributes } | ||
| }; | ||
|
|
||
| } | ||
| } |
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,21 @@ | ||
| // Copyright (c) Microsoft Corporation. All rights reserved. | ||
| // Constants.cs | ||
|
|
||
| namespace Microsoft.AutoGen.Core.Grpc; | ||
|
|
||
| public static class Constants | ||
| { | ||
| public const string DATA_CONTENT_TYPE_PROTOBUF_VALUE = "application/x-protobuf"; | ||
| public const string DATA_CONTENT_TYPE_JSON_VALUE = "application/json"; | ||
| public const string DATA_CONTENT_TYPE_TEXT_VALUE = "text/plain"; | ||
|
|
||
| public const string DATA_CONTENT_TYPE_ATTR = "datacontenttype"; | ||
| public const string DATA_SCHEMA_ATTR = "dataschema"; | ||
| public const string AGENT_SENDER_TYPE_ATTR = "agagentsendertype"; | ||
| public const string AGENT_SENDER_KEY_ATTR = "agagentsenderkey"; | ||
|
|
||
| public const string MESSAGE_KIND_ATTR = "agmsgkind"; | ||
| public const string MESSAGE_KIND_VALUE_PUBLISH = "publish"; | ||
| public const string MESSAGE_KIND_VALUE_RPC_REQUEST = "rpc_request"; | ||
| public const string MESSAGE_KIND_VALUE_RPC_RESPONSE = "rpc_response"; | ||
| } |
Oops, something went wrong.