cpr

Documentation

Overview

Package cpr contains concurrency primitives.

Index

• func Demultiplex[T any](inChs ...<-chan T) chan T
• func Get[T any](ch <-chan T, errCh <-chan error) (T, bool, error)
• func Pop[T any](ctx context.Context, ch <-chan T) (T, bool, error)
• func Push[T any](ctx context.Context, ch chan<- T, ts ...T) error
• func RunTestEngine[T any](ctx context.Context, ps Processor[T], ts ...T) ([]T, error)
• type Collector
  • func (c *Collector[T]) Sink(ctx context.Context, inCh <-chan T) error
• type Engine
  • func (eng *Engine[T]) Add(p Processor[T])
  • func (eng *Engine[T]) Process(ctx context.Context) error
• type Processor
• type Producer
  • func (p *Producer[T]) Source(ctx context.Context, outCh chan<- T) error
• type Sink
• type Source

Functions

func Demultiplex

func Demultiplex[T any](inChs ...<-chan T) chan T

Demultiplex demultiplexes the given channels.

func Get

func Get[T any](ch <-chan T, errCh <-chan error) (T, bool, error)

Get gets and returns a new T from the supplied channel. It returns a T, a boolean which indicates whether the channel is still open, or an error from the supplied errCh.

func Pop

func Pop[T any](ctx context.Context, ch <-chan T) (T, bool, error)

Pop returns a new T from the ch. It returns a boolean which indicates whether the channel is still open. The error indicates whether the context has been canceled.

func Push

func Push[T any](ctx context.Context, ch chan<- T, ts ...T) error

Push tries to push a T to the ch. The error indicates whether the context has been canceled.

func RunTestEngine

func RunTestEngine[T any](ctx context.Context, ps Processor[T], ts ...T) ([]T, error)

RunTestEngine runs the processor in a test engine.

Types

type Collector

type Collector[T any] struct {
	Result []T
}

Collector collects channel result into an array.

func (*Collector[T]) Sink

func (c *Collector[T]) Sink(ctx context.Context, inCh <-chan T) error

Sink implements Sink.

type Engine

type Engine[T any] struct {
	Source     Source[T]
	Sink       Sink[T]
	Processors []Processor[T]
}

Engine processes a pipeline.

func (*Engine[T]) Add

func (eng *Engine[T]) Add(p Processor[T])

Add adds a processor.

func (*Engine[T]) Process

func (eng *Engine[T]) Process(ctx context.Context) error

Process processes the pipeline in the engine.

type Processor

type Processor[T any] interface {
	Process(context.Context, <-chan T, chan<- T) error
}

Processor processes elements.

type Producer

type Producer[T any] struct {
	Items []T
}

Producer produces values.

func (*Producer[T]) Source

func (p *Producer[T]) Source(ctx context.Context, outCh chan<- T) error

Source implements Source.

type Sink

type Sink[T any] interface {
	Sink(context.Context, <-chan T) error
}

Sink consumes elements.

type Source

type Source[T any] interface {
	Source(context.Context, chan<- T) error
}

Source generates elements.