Documentation
¶
Overview ¶
Package stream provides canonical streaming primitives for the execution engine, including keepalive injection during recovery and idle waits.
Index ¶
- Constants
- Variables
- func ContextWithKeepaliveInterval(ctx context.Context, d time.Duration) context.Context
- func DefaultKeepaliveEvent() lipapi.Event
- func FlushSSEDataJSON(w io.Writer, fl Flusher, payload any) error
- func FlushSSEDataJoined(w io.Writer, fl Flusher, prefix, mid, suffix []byte) error
- func FlushSSEEventJSON(w io.Writer, fl Flusher, eventName string, payload any) error
- func KeepaliveIntervalFromContext(ctx context.Context) time.Duration
- func PumpSSE(ctx context.Context, w http.ResponseWriter, es lipapi.EventStream, ...) (err error)
- func WrapRecoveryKeepalive(s lipapi.EventStream) (lipapi.EventStream, error)
- func WrapRecoveryKeepaliveInterval(s lipapi.EventStream, d time.Duration) (lipapi.EventStream, error)
- type EventPump
- type Flusher
- type Keepalive
- type KeepaliveConfig
- type PendingEventQueue
Constants ¶
const DefaultRecoveryKeepaliveInterval = 12 * time.Second
DefaultRecoveryKeepaliveInterval is used while waiting on upstream streams or executor recv-phase failover (Req 5.5).
const KeepaliveEventCode = "keepalive"
KeepaliveEventCode is the WarningCode used for canonical keepalive events.
Variables ¶
var DrainPending = streampump.DrainPending
DrainPending pops every queued event in order and returns them.
var ErrNilEventStream = lipapi.ErrNilEventStream
ErrNilEventStream is returned by NewKeepalive when the inner stream is nil. It is identical to lipapi.ErrNilEventStream so callers can use errors.Is against either package.
var ErrNilKeepalive = errors.New("stream: nil Keepalive")
ErrNilKeepalive is returned by Recv when the receiver is a nil *Keepalive.
var ErrPendingQueueFull = streampump.ErrPendingQueueFull
ErrPendingQueueFull is returned when [PendingEventQueue.Push] would exceed a configured max length.
var NewPendingEventQueue = streampump.NewPendingEventQueue
NewPendingEventQueue returns a queue with the given max pending events (0 = unlimited).
Functions ¶
func ContextWithKeepaliveInterval ¶
ContextWithKeepaliveInterval attaches a recovery keepalive interval for PumpSSE.
func DefaultKeepaliveEvent ¶
DefaultKeepaliveEvent returns a standard keepalive warning event.
func FlushSSEDataJSON ¶
FlushSSEDataJSON writes one SSE data line: "data: <json>\n\n" using a pooled buffer.
func FlushSSEDataJoined ¶
FlushSSEDataJoined writes data: prefix+mid+suffix followed by "\n\n" using a pooled buffer. prefix+mid+suffix must concatenate to one valid JSON value (no leading "data: " prefix).
func FlushSSEEventJSON ¶
FlushSSEEventJSON writes one SSE record: "event: name\ndata: <json>\n\n" using a pooled buffer.
func KeepaliveIntervalFromContext ¶
KeepaliveIntervalFromContext returns the attached interval or DefaultRecoveryKeepaliveInterval.
func PumpSSE ¶
func PumpSSE(ctx context.Context, w http.ResponseWriter, es lipapi.EventStream, eofWithoutDone error, handle func(lipapi.Event) (done bool, err error)) (err error)
PumpSSE wraps es with recovery keepalive, pumps canonical events through handle until it returns done or an error, and closes es on exit (joining close errors).
func WrapRecoveryKeepalive ¶
func WrapRecoveryKeepalive(s lipapi.EventStream) (lipapi.EventStream, error)
WrapRecoveryKeepalive wraps the canonical event stream so idle reads emit protocol-neutral Warning keepalives (see DefaultKeepaliveEvent).
func WrapRecoveryKeepaliveInterval ¶
func WrapRecoveryKeepaliveInterval(s lipapi.EventStream, d time.Duration) (lipapi.EventStream, error)
WrapRecoveryKeepaliveInterval wraps s with the given idle keepalive interval.
Types ¶
type EventPump ¶
type EventPump[T any] = streampump.EventPump[T]
EventPump owns the common "pending queue before wire read" loop used by stream adapters.
type Flusher ¶
type Flusher interface {
Flush()
}
Flusher is the flush-only surface SSE writers need from an HTTP response writer; net/http.ResponseWriter satisfies it structurally.
type Keepalive ¶
type Keepalive struct {
// contains filtered or unexported fields
}
Keepalive wraps an inner EventStream and emits keepalive events at the configured interval while Recv is blocked waiting for the next real event. This prevents client-side idle timeouts during recovery waits (Req 5.5).
A single background goroutine sequentially reads from the inner stream, preventing concurrent access. Recv returns either a buffered real event or a keepalive when the inner stream hasn't produced a result within the interval. Close cancels any in-flight inner Recv (see abortRead) and closes k.done so the reader loop exits without a second goroutine.
EventStream concurrency: one goroutine calls Recv at a time. Close is safe concurrently with Recv blocked on the keepalive select or inner I/O; Close aborts the in-flight inner Recv then closes the inner stream.
func NewKeepalive ¶
func NewKeepalive(s lipapi.EventStream, cfg KeepaliveConfig) (*Keepalive, error)
NewKeepalive wraps s with keepalive injection using cfg. It returns ErrNilEventStream if s is nil.
type KeepaliveConfig ¶
type KeepaliveConfig struct {
// Interval is the idle window before a synthetic keepalive may be returned from Keepalive.Recv.
// Zero schedules the first keepalive as soon as the inner stream blocks (high churn); use a
// positive duration in production paths.
Interval time.Duration
// NewKeepalive builds the synthetic event. Nil selects DefaultKeepaliveEvent.
NewKeepalive func() lipapi.Event
// Log when set receives the rare reader-panic log when the panic cannot be sent on the result channel.
Log *slog.Logger
}
KeepaliveConfig controls keepalive event injection during idle stream waits.
type PendingEventQueue ¶
type PendingEventQueue = streampump.PendingEventQueue
PendingEventQueue buffers canonical events for adapters that translate one wire chunk into zero or more lipapi.Event values.
Source Files