datapipe

Documentation

Index

• func Broadcast[I any](input Input[I], output ...Output[I]) <-chan error
• type FlattenStrat
• type Flow
• type Input
  • func Concat[I any](input Input[I]) Input[[]I]
  • func Duplicate[T any](i Input[T], count int) []Input[T]
  • func First[I any](input Input[I]) Input[I]
  • func FlattenMap[K comparable, V any](input Input[map[K]V], strat FlattenStrat) Input[map[K]V]
  • func MapInput[I any, O any](input Input[I], mapper func(input I) (O, error)) Input[O]
  • func Merge[T any](input ...Input[T]) Input[T]
  • func NewInput[T any](i <-chan T, e <-chan error) Input[T]
  • func Scan[I any, O any](input Input[I], initialValue O, scanner func(elem I, res O) (O, error)) Input[O]
  • func Sequence[I any](input Input[[]I]) Input[I]
  • func (in Input[T]) Input() (<-chan T, <-chan error)
• type InputRaw
  • func (ir InputRaw) Input() (<-chan []byte, <-chan error)
• type Inputable
• type Output
  • func NewOutput[T any](out func(input <-chan T, err <-chan error) <-chan error) Output[T]
  • func (o Output[T]) Output(input Input[T]) <-chan error
• type OutputRaw
  • func (o OutputRaw) Output(input Input[[]byte]) <-chan error
• type Outputable
• type Pipe
  • func ConcatPipe[I any]() Pipe[I, []I]
  • func FlattenMapPipe[K comparable, V any](strat FlattenStrat) Pipe[map[K]V, map[K]V]
  • func Map[I any, O any](mapper func(input I) (O, error)) Pipe[I, O]
  • func TimeBuffered[I any](duration time.Duration) Pipe[I, []I]
  • func (p Pipe[I, O]) Pipe(input Input[I]) Input[O]
• type Pipeable

Functions

func Broadcast

func Broadcast[I any](input Input[I], output ...Output[I]) <-chan error

Types

type FlattenStrat

type FlattenStrat int

const (
	ScanStrat   FlattenStrat = 0
	ConcatStrat FlattenStrat = 1
)

type Flow

type Flow[I any, O any] struct {
	// contains filtered or unexported fields
}

type Input

type Input[T any] func() (<-chan T, <-chan error)

func Concat

func Concat[I any](input Input[I]) Input[[]I]

func Duplicate

func Duplicate[T any](i Input[T], count int) []Input[T]

func First

func First[I any](input Input[I]) Input[I]

func FlattenMap

func FlattenMap[K comparable, V any](input Input[map[K]V], strat FlattenStrat) Input[map[K]V]

func MapInput

func MapInput[I any, O any](input Input[I], mapper func(input I) (O, error)) Input[O]

func Merge

func Merge[T any](input ...Input[T]) Input[T]

func NewInput

func NewInput[T any](i <-chan T, e <-chan error) Input[T]

func Scan

func Scan[I any, O any](input Input[I], initialValue O, scanner func(elem I, res O) (O, error)) Input[O]

func Sequence

func Sequence[I any](input Input[[]I]) Input[I]

func (Input[T]) Input

func (in Input[T]) Input() (<-chan T, <-chan error)

type InputRaw

type InputRaw func() (<-chan []byte, <-chan error)

func (InputRaw) Input

func (ir InputRaw) Input() (<-chan []byte, <-chan error)

type Inputable

type Inputable[T any] interface {
	Input() (<-chan T, <-chan error)
}

type Output

type Output[T any] func(input Input[T]) <-chan error

func NewOutput

func NewOutput[T any](out func(input <-chan T, err <-chan error) <-chan error) Output[T]

func (Output[T]) Output

func (o Output[T]) Output(input Input[T]) <-chan error

type OutputRaw

type OutputRaw func(input Input[[]byte]) <-chan error

func (OutputRaw) Output

func (o OutputRaw) Output(input Input[[]byte]) <-chan error

type Outputable

type Outputable[T any] interface {
	Output(input Input[T]) <-chan error
}

type Pipe

type Pipe[I any, O any] func(input Input[I]) Input[O]

func ConcatPipe

func ConcatPipe[I any]() Pipe[I, []I]

func FlattenMapPipe

func FlattenMapPipe[K comparable, V any](strat FlattenStrat) Pipe[map[K]V, map[K]V]

func Map

func Map[I any, O any](mapper func(input I) (O, error)) Pipe[I, O]

func TimeBuffered

func TimeBuffered[I any](duration time.Duration) Pipe[I, []I]

func (Pipe[I, O]) Pipe

func (p Pipe[I, O]) Pipe(input Input[I]) Input[O]

type Pipeable

type Pipeable[I any, O any] interface {
	Pipe(input Input[I]) Input[O]
}