README
¶
Parapipe - paralleling pipeline
The library provides a zero-dependency non-blocking buffered FIFO-pipeline for structuring the code and vertically scaling your app. The main difference from a regular pipeline example you may find on the internet - pipeline executes everything on each step concurrently, yet maintaining the order. Although, this library does not use any locks or mutexes, or any other thread synchronization tools. Just pure channels.
When to use
- processed data can be divided in chunks (messages), and the flow may consist of one or more stages
- data should be processed concurrently (scaled vertically)
- the order of processing messages must be maintained
Installation
go get github.com/nazar256/parapipe
Usage
- Create a pipeline
cfg := parapipe.Config{
Concurrency: 5, // how many messages to process concurrently for each pipe
ProcessErrors: false, // messages implementing "error" interface will not be passed to subsequent workers
}
pipeline := parapipe.NewPipeline(cfg)
- Add pipes - call
Pipe()method one or more times
pipeline.Pipe(func(msg interface{}) interface{} {
typedMsg := msg.(YourInputType) // assert your type for the message
// do something and generate a new value "someValue"
return someValue
})
- Get "out" channel when all pipes are added and read results from it
for result := range pipeline.Out() {
typedResut := result.(YourResultType)
// do something with the result
}
- Close pipeline with closing it's input channel. All internal channels, goroutines, including
Out()channel will be closed in a cascade.
close(pipeline.In())
Error handling
To handle errors just return them as a result then listen to them on Out. By default, errors will not be processed by subsequent stages.
pipeline.Pipe(func(msg interface{}) interface{} {
inputValue := msg.(YourInputType) // assert your type for the message
someValue, err := someOperation(inputValue)
if err != nil {
return err // error can also be a result and can be returned from a pipeline stage (pipe)
}
return someValue
})
// ...
for result := range pipeline.Out() {
err := result.(error)
if err != nil {
// handle the error
// you may want to stop sending new values to the pipeline in your own way and do close(pipeline.In())
}
typedResut := result.(YourResultType)
// do something with the result
}
Optionally you may allow passing errors to subsequent pipes. For example, if you do not wish to stop the pipeline on errors, but rather process them in subsequent pipes.
cfg := parapipe.Config{
Concurrency: 5, // how many messages to process concurrently for each pipe
ProcessErrors: true, // messages implementing "error" interface will be passed to subsequent workers as any message
}
pipeline := parapipe.NewPipeline(cfg).
Pipe(func(msg interface{}) interface{} {
inputValue := msg.(YourInputType) // assert your type for the message
someValue, err := someOperation(inputValue)
if err != nil {
return err // error can also be a result and can be returned from a pipeline stage (pipe)
}
return someValue
}).
Pipe(func(msg interface{}) interface{} {
switch inputValue := msg.(type) {
case error:
// process error
case YourNormalExpectedType:
// process message normally
}
})
Limitations
Out()method can be used only once on each pipeline. Any subsequentPipe()call will cause panic. Though, when you need to stream values somewhere from the middle of the pipeline - just send them to your own channel.- as at the time of writing Go does not have generics, you have to assert the type for incoming messages in pipes explicitly.
Examples
AMQP middleware
Parapipe can be handful when you need to process messages in the middle concurrently, yet maintaining their order.
replies, err = amqpChannel.Consume(
q.Name, // queue
"go-amqp-example", // consumer
true, // auto-ack
false, // exclusive
false, // no-local
false, // no-wait
nil, // args
)
cfg := parapipe.Config{
Concurrency: 5, // how many messages to process concurrently for each pipe
ProcessErrors: false, // messages implementing "error" interface will not be passed to subsequent workers
}
pipeline := parapipe.NewPipeline(cfg)
go func() {
for amqpMsg := range replies {
pipeline.In() <- amqpMsg
}
close(pipeline.In())
}()
// here pipeline starts to process messages immediately (once they are sent above) even before "Out()" is called
pipeline.
Pipe(func(msg interface{}) interface{} {
event := &Event{}
_ = json.Unmarshal(msg.(amqp.Delivery).Body, event)
return event
}).
Pipe(func(msg interface{}) interface{} {
event := msg.(Event)
// validate the event
return event
}).
Pipe(func(msg interface{}) interface{} {
payload, _ := json.Marshal(msg.(Event))
// publish as new event
err = amqpChannel.Publish(
"some-exchange",
"some:routing:key",
false,
false,
amqp.Publishing{
ContentType: "application/json",
Body: payload,
})
return nil
})
Other examples
With parapipe you can:
- respond a JSON-feed as stream, retrieve, enrich and marshal each object concurrently, in maintained order and return them to the client
- fetch and merge entries from different sources as one stream
- structure your HTTP-controllers
- processing heavy files in effective way
Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Config ¶ added in v0.2.0
type Config struct {
Concurrency int // how many messages to process concurrently for each pipe
ProcessErrors bool // if false, messages implementing "error" interface will not be passed to subsequent workers
}
Config contains pipeline parameters which influence execution or behavior
type Job ¶
type Job func(msg interface{}) interface{}
Job is a short callback signature, used in pipes
type Pipeline ¶
type Pipeline struct {
// contains filtered or unexported fields
}
Pipeline executes jobs concurrently maintaining message order
func NewPipeline ¶
NewPipeline creates new Pipeline instance, "Concurrency" sets how many jobs can be executed concurrently in each pipe
func (*Pipeline) In ¶
func (p *Pipeline) In() chan<- interface{}
In returns input channel of the pipeline - the entrance of the pipeline, send there your input values
Source Files