streamchan

README

go-streaming-channel

DRY by encapsulating channel draining logic (producing and consuming items on a channel). Thanks to Go 1.18 generics it is type safe. See the tests for examples.

Documentation

Index

• func DrainStream[T any](ctx context.Context, producer Producer[T], consumer Consumer[T], ...) error
• func WithStreamMode(ctx context.Context, mode StreamMode) context.Context
• type Consumer
• type Finisher
• type Producer
• type StreamMode

Functions

func DrainStream

func DrainStream[T any](
	ctx context.Context,
	producer Producer[T],
	consumer Consumer[T],
	finisher ...Finisher,
) error

DrainStream invokes handler for every entity T read from repository stream.

Set a stream mode in context to change the handling of canceled context.

func WithStreamMode

func WithStreamMode(ctx context.Context, mode StreamMode) context.Context

WithStreamMode specifies the stream mode to use for draining.

The default is Exhaustible.

Types

type Consumer

type Consumer[T any] func(ctx context.Context, entity T) error

Consumer is invoked on each stream item encountered.

type Finisher

type Finisher func(ctx context.Context) error

Finisher is invoked after all stream items have been processed.

type Producer

type Producer[T any] func(ctx context.Context, stream chan T) error

Producer sends items on the stream channel.

The function is intended to be run as go routine. It must take care to close the channel, which signals the function is done. Closing is best deferred.

type StreamMode

type StreamMode int

const (
	// Exhaustible streams are eventually closed during program execution.
	//
	// Canceling the context is considered an error.
	Exhaustible StreamMode = iota + 1

	// Infinite streams produce values indefinitely.
	//
	// Canceling the context is not considered an error.
	Infinite
)