< Summary - AsyncResponse (Release / net8.0+net10.0 / unit+integration)

Information
Class: Microsoft.Extensions.DependencyInjection.AsyncResponseCoreServiceCollectionExtensions
Assembly: AsyncResponse.Core
File(s): /_/src/AsyncResponse.Core/ServiceCollectionExtensions.cs
Line coverage
100%
Covered lines: 156
Uncovered lines: 0
Coverable lines: 156
Total lines: 402
Line coverage: 100%
Branch coverage
95%
Covered branches: 38
Total branches: 40
Branch coverage: 95%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
AddAsyncResponse(...)100%22100%
WithDurableFlows(...)100%11100%
WithDurableFlows(...)100%88100%
WithDurableFlow(...)80%1010100%
WithScheduledFlow(...)100%1414100%
WithInMemoryDurableFlows(...)100%11100%
AddDurableFlowEngine(...)100%11100%
WithContextPropagator(...)100%11100%
WithInMemoryChannel(...)100%44100%
WithInMemoryTransport(...)100%22100%

File(s)

/_/src/AsyncResponse.Core/ServiceCollectionExtensions.cs

#LineLine coverage
 1using AsyncResponse;
 2using Microsoft.Extensions.DependencyInjection.Extensions;
 3using Microsoft.Extensions.Options;
 4using System.Diagnostics.CodeAnalysis;
 5
 6namespace Microsoft.Extensions.DependencyInjection;
 7
 8/// <summary>
 9/// Core registrations for AsyncResponse. Everything is configured through the fluent builder
 10/// returned by <see cref="AddAsyncResponse"/>: chain exactly one channel and exactly one worker
 11/// transport, and exactly one durable-flow state store.
 12/// </summary>
 13public static class AsyncResponseCoreServiceCollectionExtensions
 14{
 15    /// <summary>
 16    /// Registers the channel-agnostic AsyncResponse engine (fluent waiter builder, transport-neutral
 17    /// ingress, worker-job executor, and the recovery watchdog) and returns a builder to configure
 18    /// the rest. It deliberately registers <em>no</em> response channel: chain exactly one
 19    /// (<see cref="WithInMemoryChannel"/> or the Redis channel package's <c>WithRedisChannel</c>) and
 20    /// exactly one worker transport (<see cref="WithInMemoryTransport"/> or a broker transport
 21    /// package such as <c>WithGooglePubSubTransport</c> / <c>WithRabbitMqTransport</c>), and exactly
 22    /// one durable-flow state store (<see cref="WithInMemoryDurableFlows"/> or a provider package).
 23    /// An app that starts without any of these choices fails fast at host startup.
 24    /// </summary>
 25    public static AsyncResponseRegistrationBuilder AddAsyncResponse(
 26        this IServiceCollection services,
 27        Action<AsyncResponseOptions>? configure = null)
 28    {
 339429        services.AddOptions();
 339430        if (configure is not null)
 31        {
 223532            services.Configure(configure);
 33        }
 34
 35        // The engine's single clock. Every Core component that reads time or arms a timer resolves
 36        // this TimeProvider, so a test host (AsyncResponse.Testing's VirtualTimeProvider) can make
 37        // waits, timeouts, leases, timers, and schedules run on virtual time. Defaults to the
 38        // system clock; TryAdd lets a host (or the test harness) pre-register its own.
 339439        services.TryAddSingleton(TimeProvider.System);
 40
 41        // Channel-agnostic engine.
 339442        services.TryAddSingleton<AsyncResponseContextPropagation>();
 339443        services.TryAddSingleton<WorkerJobExecutor>();
 339444        services.TryAddSingleton<IAsyncResponseIngress, AsyncResponseIngress>();
 339445        services.TryAddSingleton<IAsyncResponseBuilder>(provider => new AsyncResponseBuilder(
 339446            provider.GetRequiredService<IAsyncResponseSubscriber>(),
 339447            provider.GetService<IWorkerTransport>(),
 339448            provider.GetService<IAsyncResponseReplyTargetProvider>(),
 339449            provider.GetRequiredService<AsyncResponseContextPropagation>(),
 339450            provider.GetService<TimeProvider>(),
 339451            // The producer-side mirror of the ingress's inbound size budget (WorkerJobTooLargeException).
 339452            provider.GetService<IOptions<AsyncResponseOptions>>()));
 53
 54        // Fail fast before background services do any real work if the required channel,
 55        // transport, and durable-flow store choices were not made explicitly. TryAddEnumerable
 56        // (keyed by implementation type) keeps a second AddAsyncResponse() call from registering a
 57        // second validator or watchdog instance.
 339458        services.TryAddEnumerable(ServiceDescriptor.Singleton<Microsoft.Extensions.Hosting.IHostedService, AsyncResponse
 59
 60        // The recovery watchdog is part of the engine and runs by default for whatever channel is
 61        // registered (scanning + liveness go through IRecoveryStateScanner / IActiveSubscriberProbe).
 339462        services.TryAddSingleton<AsyncResponseWatchdogState>();
 339463        services.TryAddEnumerable(ServiceDescriptor.Singleton<Microsoft.Extensions.Hosting.IHostedService, AsyncResponse
 64
 339465        return new AsyncResponseRegistrationBuilder(services);
 66    }
 67
 68    /// <summary>
 69    /// Uses an application-owned store for durable-flow state. The store must implement atomic
 70    /// creation, revision-checked updates, and execution leases. Use
 71    /// <see cref="WithInMemoryDurableFlows"/> explicitly for a process-local development or test
 72    /// store; <see cref="AddAsyncResponse"/> does not select one implicitly.
 73    /// </summary>
 74    public static AsyncResponseRegistrationBuilder WithDurableFlows<[DynamicallyAccessedMembers(DynamicallyAccessedMembe
 75        this AsyncResponseRegistrationBuilder builder,
 76        Action<DurableFlowOptions>? configure = null)
 77        where TFlowStateStore : class, IFlowStateStore
 85078        => builder.WithDurableFlows<TFlowStateStore, DurableFlowOptions>(configure);
 79
 80    /// <summary>
 81    /// Registers an application or provider-owned durable-flow store whose options combine the
 82    /// common <see cref="DurableFlowOptions"/> settings with store-specific settings. Provider
 83    /// packages use this overload to expose one cohesive <c>With*DurableFlows(...)</c> callback.
 84    /// </summary>
 85    public static AsyncResponseRegistrationBuilder WithDurableFlows<[DynamicallyAccessedMembers(DynamicallyAccessedMembe
 86        this AsyncResponseRegistrationBuilder builder,
 87        Action<TOptions>? configure = null)
 88        where TFlowStateStore : class, IFlowStateStore
 89        where TOptions : DurableFlowOptions
 90    {
 269291        builder.Services.AddOptions<TOptions>();
 269292        if (configure is not null)
 242793            builder.Services.Configure(configure);
 94
 95        // The engine consumes the same configured instance as the provider store, viewed through
 96        // the common base type. This keeps all durable-flow settings in one callback without a
 97        // second options object or per-execution adaptation/allocation.
 269298        builder.Services.AddSingleton<DurableFlowOptions>(provider =>
 525699            provider.GetRequiredService<Microsoft.Extensions.Options.IOptions<TOptions>>().Value);
 100
 101        // Forward the interface to the concrete registration so resolving either yields the same
 102        // instance within a scope. TryAdd lets callers (and the DurableFlows.* packages) pre-register
 103        // the concrete type with a different lifetime — e.g. singleton for stores whose dependencies
 104        // are all singletons — without this scoped default overriding it. The forward MIRRORS the
 105        // concrete registration's lifetime: a scoped factory that returns a root-owned singleton
 106        // would be captured into every flow-execution scope's disposable list, so the first scope's
 107        // disposal would dispose the store (and any connections it owns) for the whole process.
 108        // The mirror is a REGISTRATION-TIME snapshot — a concrete registration added after this
 109        // call with a different lifetime silently re-opens that hazard, so the marker carries the
 110        // snapshot and the startup validator re-checks it against the final collection.
 2692111        builder.Services.TryAddScoped<TFlowStateStore>();
 16941112        var storeLifetime = builder.Services.Last(d => d.ServiceType == typeof(TFlowStateStore)).Lifetime;
 2692113        builder.Services.Add(ServiceDescriptor.Describe(
 2692114            typeof(IFlowStateStore),
 2576115            static provider => provider.GetRequiredService<TFlowStateStore>(),
 2692116            storeLifetime));
 2692117        builder.Services.AddSingleton(new AsyncResponseDurableFlowStoreMarker(typeof(TFlowStateStore), storeLifetime, bu
 2692118        AddDurableFlowEngine(builder.Services);
 2692119        return builder;
 120    }
 121
 122    /// <summary>
 123    /// Registers a durable flow class for execution: adds it to DI (scoped, unless the app
 124    /// pre-registered it with another lifetime) and records a statically-typed execution route the
 125    /// flow executor prefers over reflection-based type-name resolution. Registration is optional
 126    /// on JIT deployments (unregistered flows resolve reflectively as before) and required for
 127    /// flows executed in trimmed/Native AOT apps, where persisted type names cannot root code.
 128    /// </summary>
 129    /// <typeparam name="TFlow">The flow class; its full name is the persisted <see cref="FlowState.FlowTypeName"/>.</ty
 130    /// <typeparam name="TInput">The flow input type, persisted as JSON with the flow state.</typeparam>
 131    public static AsyncResponseRegistrationBuilder WithDurableFlow<[DynamicallyAccessedMembers(DynamicallyAccessedMember
 132        this AsyncResponseRegistrationBuilder builder)
 133        where TFlow : class, IDurableFlow<TInput>
 134    {
 2274135        builder.Services.TryAddScoped<TFlow>();
 2274136        builder.Services.AddSingleton(new DurableFlowRegistration
 2274137        {
 2274138            FlowTypeFullName = typeof(TFlow).FullName
 2274139                ?? throw new InvalidOperationException("Durable flow classes must have a FullName."),
 2274140            InputTypeFullName = typeof(TInput).FullName
 2274141                ?? throw new InvalidOperationException("Durable flow input types must have a FullName."),
 2274142            FlowType = typeof(TFlow),
 1771143            DeserializeInput = static json => JsonSafety.SafeDeserialize<TInput>(json),
 2274144            ExecuteAsync = static (flow, context, input) =>
 1773145                ((TFlow)flow).ExecuteAsync(context, input is null ? default! : (TInput)input)
 2274146        });
 2274147        return builder;
 148    }
 149
 150    /// <summary>
 151    /// Starts <typeparamref name="TFlow"/> on a cron schedule. Occurrences carry deterministic run
 152    /// ids (<c>sched:{name}:{occurrenceUtc}</c>), so any number of replicas can run the scheduler
 153    /// and the flow store's atomic create guarantees exactly one run per occurrence — no leader
 154    /// election. Occurrences missed while no replica was up are skipped (at-most-once schedule).
 155    /// The flow type is also registered for statically-typed (AOT-safe) execution, exactly like
 156    /// <see cref="WithDurableFlow{TFlow, TInput}"/>.
 157    /// </summary>
 158    /// <param name="builder">The registration builder.</param>
 159    /// <param name="name">
 160    /// Unique schedule name, embedded in every occurrence's flow id. Keep it short and stable —
 161    /// renaming orphans no state but changes the ids future occurrences dedup on.
 162    /// </param>
 163    /// <param name="cron">
 164    /// Five-field cron expression (<c>minute hour day-of-month month day-of-week</c>); see
 165    /// <see cref="CronSchedule"/> for the supported syntax and DST semantics. Validated here, so a
 166    /// typo fails at registration rather than silently never firing.
 167    /// </param>
 168    /// <param name="input">
 169    /// Builds the flow input for an occurrence (its scheduled UTC instant is passed in). Must be
 170    /// deterministic across replicas: every replica computes the same occurrence, and the
 171    /// idempotent-start check compares the input value.
 172    /// </param>
 173    /// <param name="configure">Optional per-schedule options (time zone, enabled).</param>
 174    public static AsyncResponseRegistrationBuilder WithScheduledFlow<[DynamicallyAccessedMembers(DynamicallyAccessedMemb
 175        this AsyncResponseRegistrationBuilder builder,
 176        string name,
 177        string cron,
 178        Func<DateTimeOffset, TInput> input,
 179        Action<ScheduledFlowOptions>? configure = null)
 180        where TFlow : class, IDurableFlow<TInput>
 181    {
 40182        ArgumentException.ThrowIfNullOrWhiteSpace(name);
 40183        ArgumentException.ThrowIfNullOrWhiteSpace(cron);
 40184        ArgumentNullException.ThrowIfNull(input);
 185
 186        // Names key the deterministic occurrence ids (sched:{name}:{occurrence}); a duplicate
 187        // would make two schedules dedup against each other's runs. Fail at the registration call
 188        // site — a BackgroundService fault at startup surfaces far less directly.
 40189        if (builder.Services.Any(descriptor =>
 1208190                descriptor.ServiceType == typeof(ScheduledFlowRegistration)
 1208191                && descriptor.ImplementationInstance is ScheduledFlowRegistration existing
 1208192                && string.Equals(existing.Name, name, StringComparison.Ordinal)))
 193        {
 2194            throw new InvalidOperationException(
 2195                $"A scheduled flow named '{name}' is already registered. Schedule names key the deterministic occurrence
 2196                "(sched:{name}:{occurrence}), so each WithScheduledFlow registration needs a unique name.");
 197        }
 198
 38199        var options = new ScheduledFlowOptions();
 38200        configure?.Invoke(options);
 38201        ArgumentNullException.ThrowIfNull(options.TimeZone, $"{nameof(ScheduledFlowOptions)}.{nameof(ScheduledFlowOption
 202        // The re-drive interval arms a Task.Delay, so it gets the timer ceiling. The startup window
 203        // needs none: it is subtracted from "now" (clamped at the epoch) and bounds a look-back whose
 204        // cost follows the occurrences the probe keeps, not the window's length — see
 205        // ScheduledFlowService.RecentOccurrences. Zero disables the probe.
 38206        AsyncResponseChannelOptions.EnsureTimerBacked(options.RedriveInterval, nameof(ScheduledFlowOptions), nameof(Sche
 34207        if (options.StartupRedriveWindow < TimeSpan.Zero)
 2208            throw new ArgumentException($"{nameof(ScheduledFlowOptions)}.{nameof(ScheduledFlowOptions.StartupRedriveWind
 209
 210        // Validate the FINAL occurrence id against the WHOLE portable contract now — length,
 211        // bytes, characters, surrounding spaces — by running the id the scheduler will actually
 212        // mint through the same check the store's create uses. Duplicating one of those rules here
 213        // let a name containing '/' (or enough multi-byte characters) register cleanly and then
 214        // fail on every occurrence at 3 a.m., logged and dropped.
 32215        var occurrenceId = ScheduledFlowService.OccurrenceFlowId(name, DateTimeOffset.UnixEpoch);
 32216        if (FlowStateConcurrency.FlowIdNotPortable(occurrenceId) is { } rejection)
 217        {
 8218            throw new ArgumentException(
 8219                $"The scheduled flow name '{name}' produces occurrence ids (sched:{{name}}:{{timestamp}}) that are not p
 8220                nameof(name));
 221        }
 222
 223        // Parse in the schedule's own time zone AND probe for a real next occurrence: a well-formed
 224        // but unsatisfiable expression ("0 0 30 2 *") would otherwise register cleanly and its loop
 225        // would die at startup with one warning — the silent 3 a.m. failure this validation exists
 226        // to prevent. The runtime null-check in ScheduledFlowService stays as the backstop.
 227        // Deliberate deviation from the engine's TimeProvider seam: registration runs before any
 228        // provider (and thus any injected clock) exists, and the probe's 400-year scan (a full
 229        // Gregorian cycle — a completeness proof) means the system clock only answers "can this
 230        // expression ever fire", not "when" — the runtime loop computes actual occurrences from
 231        // the injected TimeProvider as usual.
 24232        var probe = CronSchedule.Parse(cron, options.TimeZone);
 22233        if (probe.GetNextOccurrence(TimeProvider.System.GetUtcNow()) is null)
 234        {
 2235            throw new ArgumentException(
 2236                $"The cron expression '{cron}' for scheduled flow '{name}' has no future occurrence (an unsatisfiable da
 2237                nameof(cron));
 238        }
 239
 20240        builder.WithDurableFlow<TFlow, TInput>();
 20241        builder.Services.AddSingleton(new ScheduledFlowRegistration
 20242        {
 20243            Name = name,
 20244            CronExpression = cron,
 20245            Options = options,
 20246            StartOccurrenceAsync = (flows, flowId, occurrence, cancellationToken) =>
 18247                flows.StartAsync<TFlow, TInput>(input(occurrence), flowId, cancellationToken)
 20248        });
 20249        builder.Services.TryAddEnumerable(ServiceDescriptor.Singleton<Microsoft.Extensions.Hosting.IHostedService, Sched
 20250        return builder;
 251    }
 252
 253    /// <summary>
 254    /// Uses an atomic process-local flow-state store. Intended for development, tests, and
 255    /// single-process apps; choose a DurableFlows provider package for multi-replica execution.
 256    /// </summary>
 257    public static AsyncResponseRegistrationBuilder WithInMemoryDurableFlows(
 258        this AsyncResponseRegistrationBuilder builder,
 259        Action<DurableFlowOptions>? configure = null)
 260    {
 832261        builder.Services.TryAddSingleton<InMemoryFlowStateStore>();
 832262        return builder.WithDurableFlows<InMemoryFlowStateStore>(configure);
 263    }
 264
 265    // The executor's public methods are lost-subscriber callback targets persisted by name
 266    // (RecoverAsync / FailAsync / ExecuteAsync); root them explicitly so the reflective dispatch
 267    // finds them in trimmed apps without any user action.
 268    [DynamicDependency(DynamicallyAccessedMemberTypes.PublicMethods, typeof(IDurableFlowExecutor))]
 269    [DynamicDependency(DynamicallyAccessedMemberTypes.PublicMethods, typeof(DurableFlowExecutor))]
 270    private static void AddDurableFlowEngine(IServiceCollection services)
 271    {
 272        // Observers are resolved ONCE by the executor factory below, from the root provider, and
 273        // held for the singleton executor's lifetime. A scoped/transient registration would
 274        // surface as an opaque "Cannot resolve scoped service ... from root provider" thrown per
 275        // flow job inside the transport's retry loop (ValidateOnBuild does not catch it). The
 276        // audit — evaluated by AsyncResponseStartupValidator at host start, then released — fails
 277        // with an error naming the offending registration and the fix instead, without the
 278        // executor factory rooting the service collection for the app's lifetime.
 2692279        services.TryAddSingleton(new DurableFlowObserverLifetimeAudit(services));
 4259280        services.TryAddSingleton<IDurableFlowExecutor>(provider => new DurableFlowExecutor(
 4259281            provider.GetRequiredService<IServiceScopeFactory>(),
 4259282            provider.GetRequiredService<IAsyncResponseBuilder>(),
 4259283            provider.GetRequiredService<IAsyncResponseSubscriber>(),
 4259284            provider.GetService<IRecoverableAsyncResponseSubscriber>(),
 4259285            provider.GetRequiredService<AsyncResponseContextPropagation>(),
 4259286            provider.GetRequiredService<DurableFlowOptions>(),
 4259287            provider.GetRequiredService<Microsoft.Extensions.Logging.ILogger<DurableFlowExecutor>>(),
 4259288            provider.GetServices<DurableFlowRegistration>(),
 4259289            provider.GetService<Microsoft.Extensions.Hosting.IHostApplicationLifetime>(),
 4259290            provider.GetService<TimeProvider>(),
 4259291            provider.GetServices<IDurableFlowExecutionObserver>(),
 4259292            provider.GetService<IWorkerTransport>(),
 4259293            // The registered channel's declared default waiter timeout (the startup validator
 4259294            // enforces exactly one channel); null when the channel does not declare one.
 4259295            provider.GetServices<AsyncResponseChannelMarker>()
 1567296                .Select(marker => marker.EffectiveDefaultWaitTimeout)
 5826297                .LastOrDefault(timeout => timeout is not null)));
 4281298        services.TryAddSingleton<IDurableFlows>(provider => new DurableFlowService(
 4281299            provider.GetRequiredService<IServiceScopeFactory>(),
 4281300            provider.GetRequiredService<IAsyncResponseBuilder>(),
 4281301            provider.GetRequiredService<AsyncResponseContextPropagation>(),
 4281302            provider.GetRequiredService<DurableFlowOptions>(),
 4281303            provider.GetRequiredService<Microsoft.Extensions.Logging.ILogger<DurableFlowService>>(),
 4281304            provider.GetService<TimeProvider>()));
 2692305    }
 306
 307    /// <summary>
 308    /// Registers an application <see cref="IAsyncResponseContextPropagator"/> that carries ambient
 309    /// context (trace id, principal, tenant, …) across the serialization boundary into worker jobs
 310    /// and lost-subscriber recovery callbacks. In-process hops flow ambient state automatically via
 311    /// the captured <see cref="System.Threading.ExecutionContext"/>; propagators are only needed for
 312    /// context that must survive serialization (broker-backed workers, recovery after a redeploy).
 313    /// Register one per concern (e.g. a trace propagator and a principal propagator).
 314    /// </summary>
 315    public static AsyncResponseRegistrationBuilder WithContextPropagator<[DynamicallyAccessedMembers(DynamicallyAccessed
 316        where TPropagator : class, IAsyncResponseContextPropagator
 317    {
 201318        builder.Services.TryAddEnumerable(ServiceDescriptor.Singleton<IAsyncResponseContextPropagator, TPropagator>());
 201319        return builder;
 320    }
 321
 322    /// <summary>
 323    /// Registers the process-local response channel and recovery store. Waiters, subscriptions,
 324    /// and recovery state all live in memory and disappear when the process exits — the simplest
 325    /// setup, with no durable recovery. Pair with <see cref="WithInMemoryTransport"/> for a fully
 326    /// in-memory setup including background worker jobs.
 327    /// </summary>
 328    public static AsyncResponseRegistrationBuilder WithInMemoryChannel(
 329        this AsyncResponseRegistrationBuilder builder,
 330        Action<InMemoryAsyncResponseOptions>? configure = null)
 331    {
 1342332        var services = builder.Services;
 1342333        services.AddOptions();
 1342334        if (configure is not null)
 335        {
 919336            services.Configure(configure);
 337        }
 338
 1342339        services.TryAddSingleton<InMemoryRecoveryStateStore>();
 2574340        services.TryAddSingleton<IRecoveryStateStore>(provider => provider.GetRequiredService<InMemoryRecoveryStateStore
 2134341        services.TryAddSingleton<IRecoveryStateScanner>(provider => provider.GetRequiredService<InMemoryRecoveryStateSto
 342
 1342343        services.TryAddSingleton<InMemoryAsyncResponseChannel>();
 2171344        services.TryAddSingleton<IAsyncResponsePublisher>(provider => provider.GetRequiredService<InMemoryAsyncResponseC
 1908345        services.TryAddSingleton<IRawAsyncResponsePublisher>(provider => provider.GetRequiredService<InMemoryAsyncRespon
 2086346        services.TryAddSingleton<IAsyncResponseSubscriber>(provider => provider.GetRequiredService<InMemoryAsyncResponse
 2241347        services.TryAddSingleton<IRecoverableAsyncResponseSubscriber>(provider => provider.GetRequiredService<InMemoryAs
 2182348        services.TryAddSingleton<IActiveSubscriberProbe>(provider => provider.GetRequiredService<InMemoryAsyncResponseCh
 349
 350        // Full recovery capability, exactly like the durable channel packages: the recoverable
 351        // fluent builder is available, durable flows register their lost-subscriber callbacks, and
 352        // the lost-subscriber dispatcher routes late responses through them. The store is
 353        // process-local, so this recovery spans waiter loss within one process lifetime (and the
 354        // simulated restarts of AsyncResponse.Testing) — not a real process exit.
 2229355        services.Replace(ServiceDescriptor.Singleton<IRecoverableAsyncResponseBuilder>(provider => new RecoverableAsyncR
 2229356            provider.GetRequiredService<IRecoverableAsyncResponseSubscriber>(),
 2229357            provider.GetService<IWorkerTransport>(),
 2229358            provider.GetService<IAsyncResponseReplyTargetProvider>(),
 2229359            provider.GetRequiredService<AsyncResponseContextPropagation>(),
 2229360            provider.GetService<TimeProvider>(),
 2229361            // The producer-side mirror of the ingress's inbound size budget (WorkerJobTooLargeException).
 2229362            provider.GetService<IOptions<AsyncResponseOptions>>())));
 2215363        services.Replace(ServiceDescriptor.Singleton<IAsyncResponseBuilder>(provider => provider.GetRequiredService<IRec
 364
 365        // The resolved default waiter timeout is declared through the marker so the startup
 366        // validator can require the durable-flow ledger TTL to out-live a timeout-less awaited
 367        // step, and the flow engine can extend a parked ledger by it — without either referencing
 368        // channel option types.
 1342369        services.AddSingleton(provider =>
 1342370        {
 908371            var options = provider.GetRequiredService<Microsoft.Extensions.Options.IOptions<InMemoryAsyncResponseOptions
 908372            return new AsyncResponseChannelMarker("InMemory")
 908373            {
 908374                EffectiveDefaultWaitTimeout = options.DefaultTimeout ?? options.RecoveryStateExpiry
 908375            };
 1342376        });
 1342377        return builder;
 378    }
 379
 380    /// <summary>
 381    /// Registers the in-memory (in-process) worker transport and its background consumer. Jobs run
 382    /// in the current process and survive only as long as it does — suitable for development, tests,
 383    /// and single-node deployments. Chain exactly one transport after <see cref="AddAsyncResponse"/>;
 384    /// for distributed, durable execution use a full broker-backed transport package such as
 385    /// <c>WithGooglePubSubTransport</c> or <c>WithRabbitMqTransport</c>.
 386    /// </summary>
 387    public static AsyncResponseRegistrationBuilder WithInMemoryTransport(
 388        this AsyncResponseRegistrationBuilder builder,
 389        Action<InMemoryWorkerTransportOptions>? configure = null)
 390    {
 652391        var services = builder.Services;
 652392        services.AddOptions();
 652393        if (configure is not null)
 422394            services.Configure(configure);
 652395        services.TryAddSingleton<WorkerJobExecutor>();
 652396        services.TryAddSingleton<InMemoryWorkerTransport>();
 1218397        services.TryAddSingleton<IWorkerTransport>(provider => provider.GetRequiredService<InMemoryWorkerTransport>());
 652398        services.AddHostedService<InMemoryWorkerHost>();
 652399        services.AddSingleton(new AsyncResponseTransportMarker("InMemory"));
 652400        return builder;
 401    }
 402}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

AddAsyncResponse(Microsoft.Extensions.DependencyInjection.IServiceCollection,System.Action`1<AsyncResponse.AsyncResponseOptions>)
WithDurableFlows(Microsoft.Extensions.DependencyInjection.AsyncResponseRegistrationBuilder,System.Action`1<AsyncResponse.DurableFlowOptions>)
WithDurableFlows(Microsoft.Extensions.DependencyInjection.AsyncResponseRegistrationBuilder,System.Action`1<TOptions>)
WithDurableFlow(Microsoft.Extensions.DependencyInjection.AsyncResponseRegistrationBuilder)
WithScheduledFlow(Microsoft.Extensions.DependencyInjection.AsyncResponseRegistrationBuilder,System.String,System.String,System.Func`2<System.DateTimeOffset,TInput>,System.Action`1<AsyncResponse.ScheduledFlowOptions>)
WithInMemoryDurableFlows(Microsoft.Extensions.DependencyInjection.AsyncResponseRegistrationBuilder,System.Action`1<AsyncResponse.DurableFlowOptions>)
AddDurableFlowEngine(Microsoft.Extensions.DependencyInjection.IServiceCollection)
WithContextPropagator(Microsoft.Extensions.DependencyInjection.AsyncResponseRegistrationBuilder)
WithInMemoryChannel(Microsoft.Extensions.DependencyInjection.AsyncResponseRegistrationBuilder,System.Action`1<AsyncResponse.InMemoryAsyncResponseOptions>)
WithInMemoryTransport(Microsoft.Extensions.DependencyInjection.AsyncResponseRegistrationBuilder,System.Action`1<AsyncResponse.InMemoryWorkerTransportOptions>)