Documentation
¶
Overview ¶
Package observable provides independently owned query streams. Publishing transfers immutable values unless SubjectOptions.Clone is supplied. Closing a stream never completes its shared source. No source construction starts work.
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
var ( // ErrJoinPending means stream cleanup completion is unknown or incomplete. // Retain the stream and continue Close with a fresh budget; do not dispose // dependencies beneath its producer. Continuation must not repeat side effects. ErrJoinPending = errors.New("observable cleanup join pending") // ErrClosed means the source or stream has been closed. ErrClosed = errors.New("observable is closed") // ErrConcurrentNext rejects overlapping consumption of the same stream. ErrConcurrentNext = errors.New("observable Next is already running") // ErrOverflow terminates a slow subscriber without affecting other subscribers. ErrOverflow = errors.New("observable subscriber buffer exhausted") // ErrInvalidOptions rejects invalid construction or operation arguments. ErrInvalidOptions = errors.New("invalid observable options") // ErrProducerPanic reports a recovered producer panic without exposing its value. ErrProducerPanic = errors.New("observable producer panicked") )
Functions ¶
This section is empty.
Types ¶
type ChannelFactory ¶
ChannelFactory opens a borrowed channel and returns a cleanup callback that must cancel and join its producer. The callback must support serialized continuation after a context error, ErrJoinPending or panic, without repeating disposal or side effects: initiate them at most once inside the callback. Any other returned outcome certifies producer completion, including failure. Panics leave completion unknown; Close retains PanicError diagnostics and ErrJoinPending until a later attempt certifies completion. A nonnil callback transfers ownership even when startup fails or returns an invalid channel: Open returns a nonnil stream alongside the error, and the caller must Close it with a separate cleanup budget. Arc never closes the borrowed channel. Closing the channel means normal completion; producer failures should use FromProducer.
type CurrentSource ¶
CurrentSource distinguishes absence from a current zero or nil value. A completed source may expose its last value; a failed source returns its error.
type PublishError ¶
PublishError describes partial publishing. Delivered subscribers accepted the immutable value; Overflowed subscribers terminated. Do not blindly retry.
func (*PublishError) Error ¶
func (e *PublishError) Error() string
func (*PublishError) Unwrap ¶
func (*PublishError) Unwrap() error
Unwrap permits errors.Is(err, ErrOverflow).
type Source ¶
Source opens one independently owned stream after admission. Callers must Close even after exhaustion or cancellation. Implementations must honor cancellation.
func FromChannelFactory ¶
func FromChannelFactory[T any](factory ChannelFactory[T]) Source[T]
FromChannelFactory constructs a lazy source with explicit producer cleanup.
func FromIterator ¶
FromIterator opens and consumes one iterator per stream. Returning false from yield stops iteration, so iterator cleanup must run on early cancellation. A blocking iterator must honor its supplied context; Close never detaches it.
func FromProducer ¶
FromProducer defers activation to Open. The stream owns one worker; Close cancels and joins it. The producer must honor cancellation and join any work it creates before returning. Concurrent emit calls are serialized. Values transfer immutably; a producer must not mutate a value after emit returns. Callback use after run returns is unsupported. Producer errors remain locally inspectable.
type State ¶
type State[T any] struct { // contains filtered or unexported fields }
State is a concurrent-safe subject with atomic current-value attachment/replay. Use NewState or NewPendingState. Nil and zero initial values are current values.
Example ¶
package main
import (
"context"
"fmt"
"slices"
"github.com/cratis/arc.go/observable"
)
func main() {
ctx := context.Background()
state, err := observable.NewState([]string{}, observable.SubjectOptions[[]string]{Clone: func(value []string) ([]string, error) { return slices.Clone(value), nil }})
if err != nil {
panic(err)
}
stream, err := state.Open(ctx)
if err != nil {
panic(err)
}
initial, err := stream.Next(ctx)
if err != nil {
panic(err)
}
fmt.Println(len(initial))
if err := state.Publish(ctx, []string{"Ada"}); err != nil {
panic(err)
}
current, err := stream.Next(ctx)
if err != nil {
panic(err)
}
fmt.Println(current)
if err := stream.Close(ctx); err != nil {
panic(err)
}
}
Output: 0 [Ada]
func NewPendingState ¶
func NewPendingState[T any](options SubjectOptions[T]) (*State[T], error)
NewPendingState constructs a state with no current value.
func NewState ¶
func NewState[T any](initial T, options SubjectOptions[T]) (*State[T], error)
NewState clones initial if configured. Construction performs no I/O or work.
func (*State[T]) Complete ¶
Complete retains the last current value, if any, and completes subscriptions.
func (*State[T]) Current ¶
Current returns a detached current value when configured, preserving presence.
func (*State[T]) Fail ¶
Fail prevents subsequent reads from presenting a stale successful current value.
type Stream ¶
Stream has a single consumer. Next returns io.EOF only for normal completion; zero and nil values are real emissions. Close cancels and joins owned work. Repeated Close is safe; a timed-out Close may be continued with a later budget. Context errors or ErrJoinPending mean joining is incomplete or unknown; any other returned result certifies completion, including a final failure.
type Subject ¶
type Subject[T any] struct { // contains filtered or unexported fields }
Subject is a concurrent-safe multicast source without current-value replay. Use NewSubject; the zero value is not initialized. Complete and Fail terminate subscribers, while closing one stream affects only that subscription.
func NewSubject ¶
func NewSubject[T any](options SubjectOptions[T]) (*Subject[T], error)
NewSubject constructs a subject without activating any work.
func (*Subject[T]) Fail ¶
Fail terminates subscriptions with err, discarding buffered stale values. Nil and io.EOF are invalid failures. Repeating the same terminal action is safe.
type SubjectOptions ¶
SubjectOptions bounds each subscriber's queue. Buffer zero defaults to 16; negative values are rejected. Clone must detach nested mutable values and be concurrent-safe. It runs outside framework locks. Without Clone, values must remain immutable after publishing, including after completion.
Source Files