Documentation
¶
Overview ¶
Package reactor joins one source of occurrences to one function for the process lifetime.
The package exports:
- Source, what delivers occurrences, and Func, what handles one
- Reactor, which runs one Source into one Func, and New, which returns one
- Reactor.Start, which starts the source, and Reactor.Shutdown, which drains it
- Reactor.Ready, whether the reactor runs and its source is receiving, and Reactor.Err, the channel that reports a failure while running
- Option, which configures a Reactor, with CancelAfter, which bounds how long its drain lets the handling in flight run before it cancels the handlers, and CancelWithin, which derives that bound from a coordinator's drain timeout
- Draining, which returns a handler's drain signal
- Every, a Source that delivers the time on each tick of an interval
- Wake, which returns a Waker: a Source that delivers the time on each tick of an interval and on each Waker.Nudge, coalescing the nudges that come before a delivery into one
- Gate, a quiesce gate: background work holds it with Gate.Shared, and an operation that must run with that work paused holds it with Gate.Exclusive, each acquisition honoring a context
Lifecycle ¶
A Reactor is a lifecycle component, with the Start, Shutdown, and Ready methods other infrastructure exposes, plus Reactor.Err for a failure while running. The package imports only the standard library and knows nothing of the coordinator: the composition root makes a reactor a graph node's value, and the lifecycle package's Coordinator infers from those methods that it starts with its layer, drains before the nodes it uses, reports readiness under its node's name, and ends the run on its failure.
Drain ¶
A reactor's handling outlives the run context and ends at its layer's drain: Reactor.Start and Reactor.Shutdown state the two phases, and CancelAfter bounds the first. A handler that runs in several steps, such as a loop over bounded passes, reads Draining to stop between steps once the drain begins.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func Draining ¶
Draining returns the drain signal of the reactor running the handler whose context is ctx: a channel closed once the reactor's Shutdown begins. The handling in flight is not interrupted by it; a handler that works in steps checks it between them and returns once it is closed, leaving what remains to the next start. A context no reactor gave has no signal, and Draining returns nil, which never closes.
Types ¶
type Func ¶
Func handles one occurrence. Its context carries the source's values, any deadline the source set for the occurrence, and the reactor's drain signal, which Draining reads; it is otherwise cancelled only when the drain cancels the handlers.
type Gate ¶
type Gate struct {
// contains filtered or unexported fields
}
Gate is a quiesce gate: a process's background work, such as a reactor's handler, holds it shared, and an operation that must run with that work paused holds it exclusively. Any number of shared holders may hold it at once, or one exclusive holder and no shared one. It is a readers-writer lock whose acquisitions honor a context, so a caller that gives up stops waiting, and which prefers the exclusive side: once an exclusive holder is waiting, no new shared holder is admitted, so work that holds the gate in short, repeated turns cannot starve the operation waiting for it. The zero value is an open gate, ready to use; a Gate must not be copied after first use.
func (*Gate) Exclusive ¶
Exclusive holds the gate alone, waiting until every shared holder and any exclusive one has released it; from the moment it waits, no new shared holder is admitted. It returns the release, which is safe to call more than once, or ctx's error if ctx ends first, holding nothing and admitting the shared holders it barred.
type Option ¶
type Option func(*reactorOptions)
Option configures a Reactor.
func CancelAfter ¶
CancelAfter bounds how long Shutdown lets the handling in flight run before it cancels the handlers' contexts. Shutdown then waits for them to unwind, until its own context ends. Set it below the coordinator's drain timeout, so a reactor that had to cancel its handlers says so in Run's result; the coordinator drops any error that arrives after its own deadline. Without CancelAfter, the handlers are cancelled only when Shutdown's context ends.
func CancelWithin ¶
CancelWithin sets CancelAfter for a reactor that drains under a coordinator whose drain timeout is drain, to half of it: the handlers are cancelled inside the drain timeout, and a reactor that had to cancel them reports it well before the coordinator's deadline drops the report.
type Reactor ¶
type Reactor[T any] struct { // contains filtered or unexported fields }
Reactor runs a Source into a Func. It is single use: Start once, Shutdown once.
func (*Reactor[T]) Err ¶
Err yields the error that ends the source while the reactor is running, then closes when the source returns. It makes the reactor a lifecycle.Monitored, which the coordinator watches, so a dead reactor ends the process.
func (*Reactor[T]) Ready ¶
Ready reports whether the reactor is running and its source is ready. It turns false once Shutdown begins.
func (*Reactor[T]) Shutdown ¶
Shutdown drains the reactor in two phases. It closes the drain signal (Draining), stops the source, and waits for the handling in flight; once the CancelAfter period passes, it cancels the handlers' contexts and waits for them to unwind. It returns once Receive returns, or when ctx ends, which cancels the handlers and stops waiting. The error reports a failure sent on Err that nothing read, the handlers' cancellation once the CancelAfter period passed, or ctx's error, or else Receive's error. Shutdown before Start retires the reactor, so it never starts, and a later Shutdown of a retired reactor returns nil.
type Source ¶
type Source[T any] interface { // Receive delivers each occurrence to fn until ctx ends, then waits for // the handling in flight and returns nil. It returns an error when the // source fails, or when fn's error ends it. Receive(ctx context.Context, fn Func[T]) error // Ready reports whether the source is receiving. Ready() bool }
Source delivers occurrences to a Func.
What a handler error means is the source's own rule. An interval (Every, Wake) has nothing to redeliver, so an error ends it; a subscription redelivers the occurrence instead and keeps receiving. A source may bound each occurrence with a deadline on the context it passes to fn, such as a broker's acknowledgement wait; the reactor keeps that deadline while detaching the handler from the source's own cancellation.
func Every ¶
Every returns a source that delivers the time on every tick of d. The handler runs synchronously, and a tick that comes due while it runs is dropped, so the handler never receives a stale time. A handler error ends the source, and so the reactor, which reports it on Err; a handler that tolerates a failure returns nil. Every panics if d is not positive, as time.NewTicker does.
type Waker ¶
type Waker struct {
// contains filtered or unexported fields
}
Waker is the Wake source: a Source of times with a Nudge its producers call.
func Wake ¶
Wake returns a source that delivers the time on every tick of d, as Every does, and also on a Waker.Nudge: the interval is the backstop and the nudge the prompt, for work whose producer can say that some is waiting. Nudges coalesce: any number before the next delivery produce one, and a nudge while the handler runs produces exactly one delivery after it. A tick that comes due while the handler runs is dropped, as Every drops it. A handler error ends the source, as it ends Every. Wake panics if d is not positive.
Source Files