pipe

type Builder ¶

type Builder[IMPL NodesMap] struct {
	// contains filtered or unexported fields
}

Builder provides tools and functions to create a pipeline and add nodes and node providers to it.

func NewBuilder[IMPL NodesMap](nodesMap IMPL, defaultOpts ...Option) *Builder[IMPL]

func (b *Builder[IMPL]) Build() (*Runner, error)

type Final ¶

type Final[IN any] interface {
	Receiver[IN]
}

Final nodes go at the end of the pipeline. They only can receive data from the pipeline, despite they could export that data by other means out of the pipes library.

type FinalFunc ¶

type FinalFunc[IN any] func(in <-chan IN)

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

func IgnoreFinal[IN any]() FinalFunc[IN]

type FinalProvider ¶

type FinalProvider[IN any] func() (FinalFunc[IN], error)

FinalProvider is a function that returns a FinalFunc to be used as Final node in a pipeline. It also might return an error if there is a problem with the configuration or instantiation of the function.

If both the returned function and the error are nil, the middle node will be ignored and would be equivalent to not adding it to the pipeline.

For readability, don't do:

return nil, nil

instead, use the equivalent convenience function:

return IgnoreFinal[T](), nil

type FinalPtr ¶

type FinalPtr[IMPL NodesMap, IN any] func(IMPL) *Final[IN]

FinalPtr is a function that, given a NodesMap, returns a pointer to a Final node, which is going to be used as store destination when this function is passed as argument to AddFinalProvider or AddFinal functions.

type Middle ¶

type Middle[IN, OUT any] interface {
	Final[IN]
	Start[OUT]
}

Middle nodes go in between Start, Final or other Middle nodes. They can send and receive data.

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.

func Bypass[INOUT any]() MiddleFunc[INOUT, INOUT]

type MiddleProvider ¶

type MiddleProvider[IN, OUT any] func() (MiddleFunc[IN, OUT], error)

MiddleProvider is a function that returns a MiddleFunc to be used as Middle node in a pipeline. It also might return an error if there is a problem with the configuration or instantiation of the function.

If the IN and OUT types are different, the returned function can't be nil unless an error is returned.

If the IN and OUT type is the same, and both the returned function and the error are nil, the middle node will be bypassed and would be equivalent to not adding it to the pipeline, dyrectly bypassing the connection between its Sender nodes to its Receiver nodes.

For readability, don't do:

return nil, nil

instead, use the equivalent convenience function:

return Bypass[T](), nil

type MiddlePtr ¶

type MiddlePtr[IMPL NodesMap, IN, OUT any] func(IMPL) *Middle[IN, OUT]

MiddlePtr is a function that, given a NodesMap, returns a pointer to a Middle node, which is going to be used as store destination when this function is passed as argument to AddMiddleProvider or AddMiddle functions.

type NodesMap ¶

type NodesMap interface {
	// Connect runs the code that connects the nodes of a pipeline. It is invoked
	// by the Builder before returning the pipeline Runner.
	Connect()
}

NodesMap is any data structure that stores references to the nodes of a pipeline, and specifies how to connect them by means of its Connect method. Example:

type MyPipeline struct {
	Load      pipe.Start[string]
	Transform pipe.Middle[string, string]
	Store     pipe.Final[string]
}
func (m *MyPipeline) Connect() {
	m.Load.SendTo(m.Transform)
	m.Transform.SendTo(m.Store)
}

The fields are assigned to nodes by the Builder, by means of AddStart, AddStartProvider, AddMiddle, AddMiddleProvider, AddFinal and AddFinalProvider

type Option ¶

type Option func(options *creationOptions)

Option allows overriding the default properties of the nodes and connections of a pipeline.

func ChannelBufferLen(length int) Option

type Receiver ¶

type Receiver[IN any] interface {
	// contains filtered or unexported methods
}

Receiver is any node that can receive data from another node: Middle or Final nodes

type Runner ¶

type Runner struct {
	// contains filtered or unexported fields
}

Runner stores all the configured nodes of a pipeline once their nodes are instantiated (as specified by AddStart, AddStartProvider, AddMiddle, AddMiddleProvider, AddFinal, AddFinalProvider) and connected according to the Connect function of the NodesMap that is provided to the Builder.

func (b *Runner) Done() <-chan struct{}

func (b *Runner) Start()

type Sender ¶

type Sender[OUT any] interface {
	// SendTo connects a Sender with a group of Receiver instances.
	SendTo(r ...Receiver[OUT]) // TODO: fail if there is any middle or final node not being destination of any "SendTo"
}

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

type Start ¶

type Start[OUT any] interface {
	Sender[OUT]
}

Start nodes insert data into the pipeline. They only can send data to the pipeline, despite they could acquire data by other means out of the pipes library.

type StartFunc ¶

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

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

func IgnoreStart[OUT any]() StartFunc[OUT]

type StartProvider ¶

type StartProvider[OUT any] func() (StartFunc[OUT], error)

StartProvider is a function that returns a StartFunc to be used as Start node in a pipeline. It also might return an error if there is a problem with the configuration or instantiation of the function.

If both the returned function and the error are nil, the start node will be ignored and would be equivalent to not adding it to the pipeline.

For readability, don't do:

return nil, nil

instead, use the equivalent convenience function:

return IgnoreStart[T](), nil

type StartPtr ¶

type StartPtr[IMPL NodesMap, OUT any] func(IMPL) *Start[OUT]

StartPtr is a function that, given a NodesMap, returns a pointer to a Start node, which is going to be used as store destination when this function is passed as argument to AddStartProvider or AddStart functions.