👁 Image
Mostlylucid.Ephemeral.Atoms.SignalAware 2.6.3

Mostlylucid.Ephemeral.Atoms.SignalAware

👁 NuGet

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)