| | | 1 | | using AsyncResponse; |
| | | 2 | | using Microsoft.Extensions.DependencyInjection.Extensions; |
| | | 3 | | using System.Diagnostics.CodeAnalysis; |
| | | 4 | | |
| | | 5 | | namespace Microsoft.Extensions.DependencyInjection; |
| | | 6 | | |
| | | 7 | | /// <summary> |
| | | 8 | | /// Core registrations for AsyncResponse. Everything is configured through the fluent builder |
| | | 9 | | /// returned by <see cref="AddAsyncResponse"/>: chain exactly one channel and exactly one worker |
| | | 10 | | /// transport, and exactly one durable-flow state store. |
| | | 11 | | /// </summary> |
| | | 12 | | public static class AsyncResponseCoreServiceCollectionExtensions |
| | | 13 | | { |
| | | 14 | | /// <summary> |
| | | 15 | | /// Registers the channel-agnostic AsyncResponse engine (fluent waiter builder, transport-neutral |
| | | 16 | | /// ingress, worker-job executor, and the recovery watchdog) and returns a builder to configure |
| | | 17 | | /// the rest. It deliberately registers <em>no</em> response channel: chain exactly one |
| | | 18 | | /// (<see cref="WithInMemoryChannel"/> or the Redis channel package's <c>WithRedisChannel</c>) and |
| | | 19 | | /// exactly one worker transport (<see cref="WithInMemoryTransport"/> or a broker transport |
| | | 20 | | /// package such as <c>WithGooglePubSubTransport</c> / <c>WithRabbitMqTransport</c>), and exactly |
| | | 21 | | /// one durable-flow state store (<see cref="WithInMemoryDurableFlows"/> or a provider package). |
| | | 22 | | /// An app that starts without any of these choices fails fast at host startup. |
| | | 23 | | /// </summary> |
| | | 24 | | public static AsyncResponseRegistrationBuilder AddAsyncResponse( |
| | | 25 | | this IServiceCollection services, |
| | | 26 | | Action<AsyncResponseOptions>? configure = null) |
| | | 27 | | { |
| | 3 | 28 | | services.AddOptions(); |
| | 3 | 29 | | if (configure is not null) |
| | | 30 | | { |
| | 3 | 31 | | services.Configure(configure); |
| | | 32 | | } |
| | | 33 | | |
| | | 34 | | // Channel-agnostic engine. |
| | 3 | 35 | | services.TryAddSingleton<AsyncResponseContextPropagation>(); |
| | 3 | 36 | | services.TryAddSingleton<WorkerJobExecutor>(); |
| | 3 | 37 | | services.TryAddSingleton<IAsyncResponseIngress, AsyncResponseIngress>(); |
| | 2 | 38 | | services.TryAddSingleton<IAsyncResponseBuilder>(provider => new AsyncResponseBuilder( |
| | 2 | 39 | | provider.GetRequiredService<IAsyncResponseSubscriber>(), |
| | 2 | 40 | | provider.GetService<IWorkerTransport>(), |
| | 2 | 41 | | provider.GetService<IAsyncResponseReplyTargetProvider>(), |
| | 2 | 42 | | provider.GetRequiredService<AsyncResponseContextPropagation>())); |
| | | 43 | | |
| | | 44 | | // Fail fast before background services do any real work if the required channel, |
| | | 45 | | // transport, and durable-flow store choices were not made explicitly. TryAddEnumerable |
| | | 46 | | // (keyed by implementation type) keeps a second AddAsyncResponse() call from registering a |
| | | 47 | | // second validator or watchdog instance. |
| | 3 | 48 | | services.TryAddEnumerable(ServiceDescriptor.Singleton<Microsoft.Extensions.Hosting.IHostedService, AsyncResponse |
| | | 49 | | |
| | | 50 | | // The recovery watchdog is part of the engine and runs by default for whatever channel is |
| | | 51 | | // registered (scanning + liveness go through IRecoveryStateScanner / IActiveSubscriberProbe). |
| | 3 | 52 | | services.TryAddSingleton<AsyncResponseWatchdogState>(); |
| | 3 | 53 | | services.TryAddEnumerable(ServiceDescriptor.Singleton<Microsoft.Extensions.Hosting.IHostedService, AsyncResponse |
| | | 54 | | |
| | 3 | 55 | | return new AsyncResponseRegistrationBuilder(services); |
| | | 56 | | } |
| | | 57 | | |
| | | 58 | | /// <summary> |
| | | 59 | | /// Uses an application-owned store for durable-flow state. The store must implement atomic |
| | | 60 | | /// creation, revision-checked updates, and execution leases. Use |
| | | 61 | | /// <see cref="WithInMemoryDurableFlows"/> explicitly for a process-local development or test |
| | | 62 | | /// store; <see cref="AddAsyncResponse"/> does not select one implicitly. |
| | | 63 | | /// </summary> |
| | | 64 | | public static AsyncResponseRegistrationBuilder WithDurableFlows<[DynamicallyAccessedMembers(DynamicallyAccessedMembe |
| | | 65 | | this AsyncResponseRegistrationBuilder builder, |
| | | 66 | | Action<DurableFlowOptions>? configure = null) |
| | | 67 | | where TFlowStateStore : class, IFlowStateStore |
| | 3 | 68 | | => builder.WithDurableFlows<TFlowStateStore, DurableFlowOptions>(configure); |
| | | 69 | | |
| | | 70 | | /// <summary> |
| | | 71 | | /// Registers an application or provider-owned durable-flow store whose options combine the |
| | | 72 | | /// common <see cref="DurableFlowOptions"/> settings with store-specific settings. Provider |
| | | 73 | | /// packages use this overload to expose one cohesive <c>With*DurableFlows(...)</c> callback. |
| | | 74 | | /// </summary> |
| | | 75 | | public static AsyncResponseRegistrationBuilder WithDurableFlows<[DynamicallyAccessedMembers(DynamicallyAccessedMembe |
| | | 76 | | this AsyncResponseRegistrationBuilder builder, |
| | | 77 | | Action<TOptions>? configure = null) |
| | | 78 | | where TFlowStateStore : class, IFlowStateStore |
| | | 79 | | where TOptions : DurableFlowOptions |
| | | 80 | | { |
| | 3 | 81 | | builder.Services.AddOptions<TOptions>(); |
| | 3 | 82 | | if (configure is not null) |
| | 3 | 83 | | builder.Services.Configure(configure); |
| | | 84 | | |
| | | 85 | | // The engine consumes the same configured instance as the provider store, viewed through |
| | | 86 | | // the common base type. This keeps all durable-flow settings in one callback without a |
| | | 87 | | // second options object or per-execution adaptation/allocation. |
| | 3 | 88 | | builder.Services.AddSingleton<DurableFlowOptions>(provider => |
| | 3 | 89 | | provider.GetRequiredService<Microsoft.Extensions.Options.IOptions<TOptions>>().Value); |
| | | 90 | | |
| | | 91 | | // Forward the interface to the concrete registration so resolving either yields the same |
| | | 92 | | // instance within a scope. TryAdd lets callers (and the DurableFlows.* packages) pre-register |
| | | 93 | | // the concrete type with a different lifetime — e.g. singleton for stores whose dependencies |
| | | 94 | | // are all singletons — without this scoped default overriding it. |
| | 3 | 95 | | builder.Services.TryAddScoped<TFlowStateStore>(); |
| | 3 | 96 | | builder.Services.AddScoped<IFlowStateStore>(provider => provider.GetRequiredService<TFlowStateStore>()); |
| | 3 | 97 | | builder.Services.AddSingleton(new AsyncResponseDurableFlowStoreMarker(typeof(TFlowStateStore))); |
| | 3 | 98 | | AddDurableFlowEngine(builder.Services); |
| | 3 | 99 | | return builder; |
| | | 100 | | } |
| | | 101 | | |
| | | 102 | | /// <summary> |
| | | 103 | | /// Registers a durable flow class for execution: adds it to DI (scoped, unless the app |
| | | 104 | | /// pre-registered it with another lifetime) and records a statically-typed execution route the |
| | | 105 | | /// flow executor prefers over reflection-based type-name resolution. Registration is optional |
| | | 106 | | /// on JIT deployments (unregistered flows resolve reflectively as before) and required for |
| | | 107 | | /// flows executed in trimmed/Native AOT apps, where persisted type names cannot root code. |
| | | 108 | | /// </summary> |
| | | 109 | | /// <typeparam name="TFlow">The flow class; its full name is the persisted <see cref="FlowState.FlowTypeName"/>.</ty |
| | | 110 | | /// <typeparam name="TInput">The flow input type, persisted as JSON with the flow state.</typeparam> |
| | | 111 | | public static AsyncResponseRegistrationBuilder WithDurableFlow<[DynamicallyAccessedMembers(DynamicallyAccessedMember |
| | | 112 | | this AsyncResponseRegistrationBuilder builder) |
| | | 113 | | where TFlow : class, IDurableFlow<TInput> |
| | | 114 | | { |
| | 3 | 115 | | builder.Services.TryAddScoped<TFlow>(); |
| | 3 | 116 | | builder.Services.AddSingleton(new DurableFlowRegistration |
| | 3 | 117 | | { |
| | 3 | 118 | | FlowTypeFullName = typeof(TFlow).FullName |
| | 3 | 119 | | ?? throw new InvalidOperationException("Durable flow classes must have a FullName."), |
| | 3 | 120 | | FlowType = typeof(TFlow), |
| | 3 | 121 | | DeserializeInput = static json => JsonSafety.SafeDeserialize<TInput>(json), |
| | 3 | 122 | | ExecuteAsync = static (flow, context, input) => |
| | 3 | 123 | | ((TFlow)flow).ExecuteAsync(context, input is null ? default! : (TInput)input) |
| | 3 | 124 | | }); |
| | 3 | 125 | | return builder; |
| | | 126 | | } |
| | | 127 | | |
| | | 128 | | /// <summary> |
| | | 129 | | /// Uses an atomic process-local flow-state store. Intended for development, tests, and |
| | | 130 | | /// single-process apps; choose a DurableFlows provider package for multi-replica execution. |
| | | 131 | | /// </summary> |
| | | 132 | | public static AsyncResponseRegistrationBuilder WithInMemoryDurableFlows( |
| | | 133 | | this AsyncResponseRegistrationBuilder builder, |
| | | 134 | | Action<DurableFlowOptions>? configure = null) |
| | | 135 | | { |
| | 3 | 136 | | builder.Services.TryAddSingleton<InMemoryFlowStateStore>(); |
| | 3 | 137 | | return builder.WithDurableFlows<InMemoryFlowStateStore>(configure); |
| | | 138 | | } |
| | | 139 | | |
| | | 140 | | // The executor's public methods are lost-subscriber callback targets persisted by name |
| | | 141 | | // (RecoverAsync / FailAsync / ExecuteAsync); root them explicitly so the reflective dispatch |
| | | 142 | | // finds them in trimmed apps without any user action. |
| | | 143 | | [DynamicDependency(DynamicallyAccessedMemberTypes.PublicMethods, typeof(IDurableFlowExecutor))] |
| | | 144 | | [DynamicDependency(DynamicallyAccessedMemberTypes.PublicMethods, typeof(DurableFlowExecutor))] |
| | | 145 | | private static void AddDurableFlowEngine(IServiceCollection services) |
| | | 146 | | { |
| | 3 | 147 | | services.TryAddSingleton<IDurableFlowExecutor>(provider => new DurableFlowExecutor( |
| | 3 | 148 | | provider.GetRequiredService<IServiceScopeFactory>(), |
| | 3 | 149 | | provider.GetRequiredService<IAsyncResponseBuilder>(), |
| | 3 | 150 | | provider.GetRequiredService<IAsyncResponseSubscriber>(), |
| | 3 | 151 | | provider.GetService<IRecoverableAsyncResponseSubscriber>(), |
| | 3 | 152 | | provider.GetRequiredService<AsyncResponseContextPropagation>(), |
| | 3 | 153 | | provider.GetRequiredService<DurableFlowOptions>(), |
| | 3 | 154 | | provider.GetRequiredService<Microsoft.Extensions.Logging.ILogger<DurableFlowExecutor>>(), |
| | 3 | 155 | | provider.GetServices<DurableFlowRegistration>())); |
| | 3 | 156 | | services.TryAddSingleton<IDurableFlows>(provider => new DurableFlowService( |
| | 3 | 157 | | provider.GetRequiredService<IServiceScopeFactory>(), |
| | 3 | 158 | | provider.GetRequiredService<IAsyncResponseBuilder>(), |
| | 3 | 159 | | provider.GetRequiredService<AsyncResponseContextPropagation>(), |
| | 3 | 160 | | provider.GetRequiredService<DurableFlowOptions>(), |
| | 3 | 161 | | provider.GetRequiredService<Microsoft.Extensions.Logging.ILogger<DurableFlowService>>())); |
| | 3 | 162 | | } |
| | | 163 | | |
| | | 164 | | /// <summary> |
| | | 165 | | /// Registers an application <see cref="IAsyncResponseContextPropagator"/> that carries ambient |
| | | 166 | | /// context (trace id, principal, tenant, …) across the serialization boundary into worker jobs |
| | | 167 | | /// and lost-subscriber recovery callbacks. In-process hops flow ambient state automatically via |
| | | 168 | | /// the captured <see cref="System.Threading.ExecutionContext"/>; propagators are only needed for |
| | | 169 | | /// context that must survive serialization (broker-backed workers, recovery after a redeploy). |
| | | 170 | | /// Register one per concern (e.g. a trace propagator and a principal propagator). |
| | | 171 | | /// </summary> |
| | | 172 | | public static AsyncResponseRegistrationBuilder WithContextPropagator<[DynamicallyAccessedMembers(DynamicallyAccessed |
| | | 173 | | where TPropagator : class, IAsyncResponseContextPropagator |
| | | 174 | | { |
| | 3 | 175 | | builder.Services.TryAddEnumerable(ServiceDescriptor.Singleton<IAsyncResponseContextPropagator, TPropagator>()); |
| | 3 | 176 | | return builder; |
| | | 177 | | } |
| | | 178 | | |
| | | 179 | | /// <summary> |
| | | 180 | | /// Registers the process-local response channel and recovery store. Waiters, subscriptions, |
| | | 181 | | /// and recovery state all live in memory and disappear when the process exits — the simplest |
| | | 182 | | /// setup, with no durable recovery. Pair with <see cref="WithInMemoryTransport"/> for a fully |
| | | 183 | | /// in-memory setup including background worker jobs. |
| | | 184 | | /// </summary> |
| | | 185 | | public static AsyncResponseRegistrationBuilder WithInMemoryChannel( |
| | | 186 | | this AsyncResponseRegistrationBuilder builder, |
| | | 187 | | Action<InMemoryAsyncResponseOptions>? configure = null) |
| | | 188 | | { |
| | 2 | 189 | | var services = builder.Services; |
| | 2 | 190 | | services.AddOptions(); |
| | 2 | 191 | | if (configure is not null) |
| | | 192 | | { |
| | 2 | 193 | | services.Configure(configure); |
| | | 194 | | } |
| | | 195 | | |
| | 2 | 196 | | services.TryAddSingleton<InMemoryRecoveryStateStore>(); |
| | 2 | 197 | | services.TryAddSingleton<IRecoveryStateStore>(provider => provider.GetRequiredService<InMemoryRecoveryStateStore |
| | 2 | 198 | | services.TryAddSingleton<IRecoveryStateScanner>(provider => provider.GetRequiredService<InMemoryRecoveryStateSto |
| | | 199 | | |
| | 2 | 200 | | services.TryAddSingleton<InMemoryAsyncResponseChannel>(); |
| | 2 | 201 | | services.TryAddSingleton<IAsyncResponsePublisher>(provider => provider.GetRequiredService<InMemoryAsyncResponseC |
| | 2 | 202 | | services.TryAddSingleton<IRawAsyncResponsePublisher>(provider => provider.GetRequiredService<InMemoryAsyncRespon |
| | 2 | 203 | | services.TryAddSingleton<IAsyncResponseSubscriber>(provider => provider.GetRequiredService<InMemoryAsyncResponse |
| | 2 | 204 | | services.TryAddSingleton<IActiveSubscriberProbe>(provider => provider.GetRequiredService<InMemoryAsyncResponseCh |
| | | 205 | | |
| | 2 | 206 | | services.AddSingleton(new AsyncResponseChannelMarker("InMemory")); |
| | 2 | 207 | | return builder; |
| | | 208 | | } |
| | | 209 | | |
| | | 210 | | /// <summary> |
| | | 211 | | /// Registers the in-memory (in-process) worker transport and its background consumer. Jobs run |
| | | 212 | | /// in the current process and survive only as long as it does — suitable for development, tests, |
| | | 213 | | /// and single-node deployments. Chain exactly one transport after <see cref="AddAsyncResponse"/>; |
| | | 214 | | /// for distributed, durable execution use a full broker-backed transport package such as |
| | | 215 | | /// <c>WithGooglePubSubTransport</c> or <c>WithRabbitMqTransport</c>. |
| | | 216 | | /// </summary> |
| | | 217 | | public static AsyncResponseRegistrationBuilder WithInMemoryTransport( |
| | | 218 | | this AsyncResponseRegistrationBuilder builder, |
| | | 219 | | Action<InMemoryWorkerTransportOptions>? configure = null) |
| | | 220 | | { |
| | 2 | 221 | | var services = builder.Services; |
| | 2 | 222 | | services.AddOptions(); |
| | 2 | 223 | | if (configure is not null) |
| | 2 | 224 | | services.Configure(configure); |
| | 2 | 225 | | services.TryAddSingleton<WorkerJobExecutor>(); |
| | 2 | 226 | | services.TryAddSingleton<InMemoryWorkerTransport>(); |
| | 2 | 227 | | services.TryAddSingleton<IWorkerTransport>(provider => provider.GetRequiredService<InMemoryWorkerTransport>()); |
| | 2 | 228 | | services.AddHostedService<InMemoryWorkerHost>(); |
| | 2 | 229 | | services.AddSingleton(new AsyncResponseTransportMarker("InMemory")); |
| | 2 | 230 | | return builder; |
| | | 231 | | } |
| | | 232 | | } |