stream

Documentation

Overview

Package stream provides cancellation-aware, bounded streams used throughout the SDK. The core channel uses no helper goroutine and allocates its ring once.

Index

• Variables
• type Channel
  • func NewChannel[T any](capacity int) *Channel[T]
  • func (c *Channel[T]) Abort(err error) error
  • func (c *Channel[T]) Cap() int
  • func (c *Channel[T]) Close() error
  • func (c *Channel[T]) Closed() bool
  • func (c *Channel[T]) Len() int
  • func (c *Channel[T]) Range(ctx context.Context) iter.Seq2[T, error]
  • func (c *Channel[T]) Recv(ctx context.Context) (T, error)
  • func (c *Channel[T]) Send(ctx context.Context, value T) error
  • func (c *Channel[T]) TryRecv() (value T, ok bool, err error)
  • func (c *Channel[T]) TrySend(value T) bool
• type Deferred
  • func NewDeferred[T any](capacity int) *Deferred[T]
  • func (d *Deferred[T]) Detach(cause error)
  • func (d *Deferred[T]) Recv(ctx context.Context) (T, error)
  • func (d *Deferred[T]) SetSource(parent context.Context, source Reader[T]) error
  • func (d *Deferred[T]) Wait()
• type MultiInput
  • func NewMultiInput[T any](parent context.Context, capacity int, onError func(string, error)) *MultiInput[T]
  • func (m *MultiInput[T]) Add(source Reader[T]) (string, error)
  • func (m *MultiInput[T]) Close() error
  • func (m *MultiInput[T]) InputCount() int
  • func (m *MultiInput[T]) Recv(ctx context.Context) (T, error)
  • func (m *MultiInput[T]) Remove(id string)
• type Reader
• type SliceReader
  • func FromSlice[T any](values []T) *SliceReader[T]
  • func (r *SliceReader[T]) Recv(ctx context.Context) (T, error)
• type Writer

Variables

var ErrClosed = errors.New("stream closed")

var ErrSourceAlreadySet = errors.New("stream source already set")

Types

type Channel

type Channel[T any] struct {
	// contains filtered or unexported fields
}

Channel is a bounded MPMC stream with explicit error termination.

func NewChannel

func NewChannel[T any](capacity int) *Channel[T]

func (*Channel[T]) Abort

func (c *Channel[T]) Abort(err error) error

func (*Channel[T]) Cap

func (c *Channel[T]) Cap() int

func (*Channel[T]) Close

func (c *Channel[T]) Close() error

func (*Channel[T]) Closed

func (c *Channel[T]) Closed() bool

func (*Channel[T]) Len

func (c *Channel[T]) Len() int

func (*Channel[T]) Range

func (c *Channel[T]) Range(ctx context.Context) iter.Seq2[T, error]

func (*Channel[T]) Recv

func (c *Channel[T]) Recv(ctx context.Context) (T, error)

func (*Channel[T]) Send

func (c *Channel[T]) Send(ctx context.Context, value T) error

func (*Channel[T]) TryRecv

func (c *Channel[T]) TryRecv() (value T, ok bool, err error)

func (*Channel[T]) TrySend

func (c *Channel[T]) TrySend(value T) bool

type Deferred

type Deferred[T any] struct {
	// contains filtered or unexported fields
}

Deferred presents a Reader before its actual source is available.

func NewDeferred

func NewDeferred[T any](capacity int) *Deferred[T]

func (*Deferred[T]) Detach

func (d *Deferred[T]) Detach(cause error)

func (*Deferred[T]) Recv

func (d *Deferred[T]) Recv(ctx context.Context) (T, error)

func (*Deferred[T]) SetSource

func (d *Deferred[T]) SetSource(parent context.Context, source Reader[T]) error

func (*Deferred[T]) Wait

func (d *Deferred[T]) Wait()

type MultiInput

type MultiInput[T any] struct {
	// contains filtered or unexported fields
}

MultiInput fans dynamically added readers into one bounded output. A failed input is removed without terminating other inputs.

func NewMultiInput

func NewMultiInput[T any](parent context.Context, capacity int, onError func(string, error)) *MultiInput[T]

func (*MultiInput[T]) Add

func (m *MultiInput[T]) Add(source Reader[T]) (string, error)

func (*MultiInput[T]) Close

func (m *MultiInput[T]) Close() error

func (*MultiInput[T]) InputCount

func (m *MultiInput[T]) InputCount() int

func (*MultiInput[T]) Recv

func (m *MultiInput[T]) Recv(ctx context.Context) (T, error)

func (*MultiInput[T]) Remove

func (m *MultiInput[T]) Remove(id string)

type Reader

type Reader[T any] interface {
	Recv(context.Context) (T, error)
}

type SliceReader

type SliceReader[T any] struct {
	// contains filtered or unexported fields
}

func FromSlice

func FromSlice[T any](values []T) *SliceReader[T]

func (*SliceReader[T]) Recv

func (r *SliceReader[T]) Recv(ctx context.Context) (T, error)

type Writer

type Writer[T any] interface {
	Send(context.Context, T) error
	Close() error
	Abort(error) error
}