steps

Documentation

Index

• Variables
• type Batcher
  • func (s Batcher) Do(ctx context.Context, inCh chan interface{}, errCh chan error) chan interface{}
• type MapFn
• type Mapper
  • func (m Mapper) Do(ctx context.Context, inCh chan any, errCh chan error) chan any
• type Receiver
• type ReduceFn
• type Reducer
  • func (s Reducer) Do(ctx context.Context, inCh chan interface{}, errCh chan error) chan interface{}
• type SubscriberReceiver
  • func (sr SubscriberReceiver) Do(ctx context.Context, _ chan interface{}, errCh chan error) chan interface{}

Variables

var ErrNonListValue = errors.New("cannot map data that isn't an Slice")

ErrNonListValue is returned when a value that is not a list flows through the pipe at Reduce step.

Types

type Batcher

type Batcher struct {
	BatchSize int
	Timeout   time.Duration
	ItemType  reflect.Type
	// contains filtered or unexported fields
}

Batcher is a Pubsub's subscriber pipeline step that accumulates messages in batches. It is useful in situations where there is a bunch of unitary messages that should be grouped to reduce systems internal I/Os, improving its performance and scale capabilities.

func (Batcher) Do

func (s Batcher) Do(ctx context.Context, inCh chan interface{}, errCh chan error) chan interface{}

Do executes a Batch pipeline.

type MapFn

type MapFn func(any) (any, error)

MapFn is the function that modifies the data that passes through the pipeline.

type Mapper

type Mapper struct {
	MapFn MapFn
}

Mapper is a pipeline step that modifies each record that passes through the pipeline.

func (Mapper) Do

func (m Mapper) Do(ctx context.Context, inCh chan any, errCh chan error) chan any

Do executes a Map pipeline.

type Receiver

type Receiver interface {
	// Receive calls f with the outstanding messages from the subscription.
	// It blocks until ctx is done, or the service returns a non-retryable error.
	Receive(ctx context.Context, f func(context.Context, *pubsub.Message)) error
}

Receiver defines something that knows how to receive Pubsub messages just like a Pubsub Subscription would.

type ReduceFn

type ReduceFn func(state interface{}, item interface{}, idx int) (newState interface{}, err error)

ReduceFn is the function that aggregates the data that passes through the pipeline into one final state.

type Reducer

type Reducer struct {
	ReduceFn     ReduceFn
	InitialState func() interface{}
	// contains filtered or unexported fields
}

Reducer is a Pubsub's subscriber pipeline step that aggregates a list of incoming pipe records into one.

func (Reducer) Do

func (s Reducer) Do(ctx context.Context, inCh chan interface{}, errCh chan error) chan interface{}

Do executes a Reduce pipeline.

type SubscriberReceiver

type SubscriberReceiver struct {
	Subscription Receiver
}

SubscriberReceiver is the first step of the pipeline which is responsible for reading the message from pubsub subscription.

func (SubscriberReceiver) Do

func (sr SubscriberReceiver) Do(ctx context.Context, _ chan interface{}, errCh chan error) chan interface{}

Do executes the messageReceiver pipeline step.