Mostlylucid.Ephemeral.Atoms.SignalAware 2.10.0

Mostlylucid.Ephemeral.Atoms.SignalAware

Pause or cancel intake based on ambient signals. Circuit-breaker and backpressure patterns.

dotnet add package mostlylucid.ephemeral.atoms.signalaware

Quick Start

using Mostlylucid.Ephemeral.Atoms.SignalAware;

var sink = new SignalSink();

await using var atom = new SignalAwareAtom<WorkItem>(
    async (item, ct) => await ProcessAsync(item, ct),
    cancelOn: new HashSet<string> { "shutdown" },
    deferOn: new HashSet<string> { "backpressure" },
    signals: sink);

await atom.EnqueueAsync(item);   // Normal processing

sink.Raise("backpressure");       // New items defer
sink.Raise("shutdown");           // New items rejected (-1)

await atom.DrainAsync();

All Options

new SignalAwareAtom<T>(
    // Required: async work body
    body: async (item, ct) => await ProcessAsync(item, ct),

    // Signals that reject items (returns -1)
    // Supports glob: "error.*", "circuit-*"
    // Default: null
    cancelOn: new HashSet<string> { "shutdown", "circuit-open" },

    // Signals that delay items until cleared
    // Supports glob patterns
    // Default: null
    deferOn: new HashSet<string> { "backpressure.*", "rate-limited" },

    // Recheck interval when deferring
    // Default: 100ms
    deferInterval: TimeSpan.FromMilliseconds(100),

    // Max defer attempts before running anyway
    // Default: 50
    maxDeferAttempts: 50,

    // Shared signal sink
    // Default: null
    signals: sharedSink,

    // Max concurrent operations
    // Default: Environment.ProcessorCount
    maxConcurrency: 8
)

API Reference

// Enqueue (returns -1 if canceled by signal)
ValueTask<long> id = await atom.EnqueueAsync(item, ct);

// Raise ambient signal
atom.Raise("backpressure");

// Drain and stats
await atom.DrainAsync(ct);
var snapshot = atom.Snapshot();
var (pending, active, completed, failed) = atom.Stats();

await atom.DisposeAsync();

Example: Circuit Breaker

var sink = new SignalSink();

await using var atom = new SignalAwareAtom<ApiRequest>(
    async (req, ct) =>
    {
        try { await CallApi(req, ct); }
        catch { sink.Raise("api.failure"); throw; }
    },
    cancelOn: new HashSet<string> { "circuit-open" },
    signals: sink);

// Monitor and open circuit
Task.Run(async () =>
{
    while (true)
    {
        if (sink.Sense(s => s.Signal == "api.failure").Count() > 5)
            sink.Raise("circuit-open");
        await Task.Delay(1000);
    }
});

Example: Backpressure

var sink = new SignalSink();

await using var atom = new SignalAwareAtom<WorkItem>(
    async (item, ct) => await SlowProcess(item, ct),
    deferOn: new HashSet<string> { "backpressure.*" },
    signals: sink);

sink.Raise("backpressure.downstream");  // Items wait
sink.Retract("backpressure.downstream"); // Items resume

Related Packages

License

Unlicense (public domain)