cpr

Documentation

Overview

Package cpr contains concurrency primitives.

Index

• func Consume[T any](ctx context.Context, ch <-chan T, f func(T) error) error
• func Demultiplex[T any](inChs ...<-chan T) chan T
• func ForAll[T any](ts []T, f func(T)) func()
• func Parallel(fs ...func()) func()
• 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 Pipeline
  • func Connect[T any](src Source[T], snk Sink[T]) Pipeline
• type Processor
• type Producer
  • func (c *Producer[T]) Source(ctx context.Context, outCh chan<- T) error
• type Sink
• type Source
  • func Compose[T any](s Source[T], p Processor[T]) Source[T]

Functions

func Consume

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

func Demultiplex

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

Demultiplex demultiplexes the given channels.

func ForAll

func ForAll[T any](ts []T, f func(T)) func()

func Parallel

func Parallel(fs ...func()) func()

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 Pipeline

type Pipeline interface {
	Process(context.Context) error
}

Pipeline represents a computation.

func Connect

func Connect[T any](src Source[T], snk Sink[T]) Pipeline

Connect connects a source and a sink.

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 (c *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.

func Compose

func Compose[T any](s Source[T], p Processor[T]) Source[T]

Compose composes a source and a processor.