node

Documentation

Overview

Package node provides functionalities to create nodes and interconnect them. A Node is a function container that can be connected via channels to other nodes. A node can send data to multiple nodes, and receive data from multiple nodes.

Index

• type Middle
  • func AsMiddle[IN, OUT any](fun MiddleFunc[IN, OUT], opts ...Option) *Middle[IN, OUT]
  • func (m *Middle[IN, OUT]) InType() reflect.Type
  • func (m *Middle[IN, OUT]) OutType() reflect.Type
  • func (s *Middle[IN, OUT]) SendsTo(outputs ...Receiver[OUT])
• type MiddleFunc
• type Option
  • func ChannelBufferLen(length int) Option
• type Receiver
• type Sender
• type Start
  • func AsStart[OUT any](funs ...StartFunc[OUT]) *Start[OUT]
  • func AsStartCtx[OUT any](funs ...StartFuncCtx[OUT]) *Start[OUT]
  • func (s *Start[OUT]) OutType() reflect.Type
  • func (s *Start[OUT]) SendsTo(outputs ...Receiver[OUT])
  • func (i *Start[OUT]) Start()
  • func (i *Start[OUT]) StartCtx(ctx context.Context)
• type StartFunc
• type StartFuncCtx
• type Terminal
  • func AsTerminal[IN any](fun TerminalFunc[IN], opts ...Option) *Terminal[IN]
  • func (t *Terminal[IN]) Done() <-chan struct{}
  • func (m *Terminal[IN]) InType() reflect.Type
• type TerminalFunc

Types

type Middle

type Middle[IN, OUT any] struct {
	// contains filtered or unexported fields
}

Middle is any intermediate node that receives data from another node, processes/filters it, and forwards the data to another node. An Middle node must have at least one output node.

func AsMiddle

func AsMiddle[IN, OUT any](fun MiddleFunc[IN, OUT], opts ...Option) *Middle[IN, OUT]

AsMiddle wraps an MiddleFunc into an Middle node.

func (*Middle[IN, OUT]) InType

func (m *Middle[IN, OUT]) InType() reflect.Type

func (*Middle[IN, OUT]) OutType

func (m *Middle[IN, OUT]) OutType() reflect.Type

func (*Middle[IN, OUT]) SendsTo

func (s *Middle[IN, OUT]) SendsTo(outputs ...Receiver[OUT])

type MiddleFunc

type MiddleFunc[IN, OUT any] func(in <-chan IN, out chan<- OUT)

MiddleFunc is a function that receives a readable channel as first argument, and a writable channel as second argument. It must process the inputs from the input channel until it's closed.

type Option

type Option func(options *creationOptions)

Option allows overriding the default values of node instantiation

func ChannelBufferLen

func ChannelBufferLen(length int) Option

ChannelBufferLen is a node.Option that allows specifying the length of the input channels for a given node. The default value is 0, which means that the channels are unbuffered.

type Receiver

type Receiver[IN any] interface {

	// InType returns the inner type of the Receiver's input channel
	InType() reflect.Type
	// contains filtered or unexported methods
}

Receiver is any node that can receive data from another node: node.Middle and node.Terminal

type Sender

type Sender[OUT any] interface {
	// SendsTo connect a sender with a group of receivers
	SendsTo(...Receiver[OUT])
	// OutType returns the inner type of the Sender's output channel
	OutType() reflect.Type
}

Sender is any node that can send data to another node: node.Start and node.Middle

type Start

type Start[OUT any] struct {
	// contains filtered or unexported fields
}

Start nodes are the starting points of a graph. This is, all the nodes that bring information from outside the graph: e.g. because they generate them or because they acquire them from an external source like a Web Service. A graph must have at least one Start node. An Start node must have at least one output node.

func AsStart

func AsStart[OUT any](funs ...StartFunc[OUT]) *Start[OUT]

AsStart wraps a group of StartFunc with the same signature into a Start node. Deprecated in favor of AsStartCtx

func AsStartCtx

func AsStartCtx[OUT any](funs ...StartFuncCtx[OUT]) *Start[OUT]

AsStartCtx wraps a group of StartFuncCtx into a Start node.

func (*Start[OUT]) OutType

func (s *Start[OUT]) OutType() reflect.Type

OutType is deprecated. It will be removed in future versions.

func (*Start[OUT]) SendsTo

func (s *Start[OUT]) SendsTo(outputs ...Receiver[OUT])

func (*Start[OUT]) Start

func (i *Start[OUT]) Start()

Start the function wrapped in the Start node. Either this method or StartCtx should be invoked for all the start nodes of the same graph, so the graph can properly start and finish.

func (*Start[OUT]) StartCtx

func (i *Start[OUT]) StartCtx(ctx context.Context)

StartCtx starts the function wrapped in the Start node, allow passing a context that can be used by the wrapped function. Either this method or Start should be invoked for all the start nodes of the same graph, so the graph can properly start and finish.

type StartFunc

type StartFunc[OUT any] func(out chan<- OUT)

StartFunc is a function that receives a writable channel as unique argument, and sends value to that channel during an indefinite amount of time.

type StartFuncCtx

type StartFuncCtx[OUT any] func(ctx context.Context, out chan<- OUT)

StartFuncCtx is a StartFunc that also receives a context as a first argument. If the passed context is cancelled via the ctx.Done() function, the implementer function should end, so the cancel will be propagated to the later nodes.

type Terminal

type Terminal[IN any] struct {
	// contains filtered or unexported fields
}

Terminal is any node that receives data from another node and does not forward it to another node, but can process it and send the results to outside the graph (e.g. memory, storage, web...)

func AsTerminal

func AsTerminal[IN any](fun TerminalFunc[IN], opts ...Option) *Terminal[IN]

AsTerminal wraps a TerminalFunc into a Terminal node.

func (*Terminal[IN]) Done

func (t *Terminal[IN]) Done() <-chan struct{}

Done returns a channel that is closed when the Terminal node has ended its processing. This is, when all its inputs have been also closed. Waiting for all the Terminal nodes to finish allows blocking the execution until all the data in the graph has been processed and all the previous stages have ended

func (*Terminal[IN]) InType

func (m *Terminal[IN]) InType() reflect.Type

type TerminalFunc

type TerminalFunc[IN any] func(out <-chan IN)

TerminalFunc is a function that receives a readable channel as unique argument. It must process the inputs from the input channel until it's closed.