parapipe

README

Parapipe - paralleling pipeline

The library provides a buffered 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 channels and goroutines.

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

1. Create a pipeline

concurrencyFactor := 5 // how many callbacks can be executed concurrently for each pipe
pipeline := parapipe.NewPipeline(concurrencyFactor)

1. Add pipes - call Pipe() method one or more times

pipeline.Pipe(func(msg interface{}) interface{} {
    typedMsg := msg.(YourInputType)     // cast message to your type
    // do something and generate a new value "someValue"
    return someValue
})

1. Get "out" channel when all pipes are added and read results from it

for result := range pipeline.Out() {
    // do something with the result, yout may have to type cast it
}

1. Close pipeline with closing it's input channel. All internal channels, goroutines, including Out() channel will be closed in a cascade.

close(pipeline.In())

Limitations

• Out() method can be used only once on each pipeline. Any subsequent Pipe() 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.
• while in the time of writing Go does not have generics, you have to cast incoming messages in pipes explicitly.

It's best shown in examples. Every box executed in each pipe in parallel.

Examples

AMQP middleware

Parapipe can be handful when you need to process messagess 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
)

concurrency := 4
pipeSource := interface{}(replies).(<-chan interface{})
pipeline := parapipe.NewPipeline(concurrency)

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

Documentation

Index

• type Job
• type Pipeline
  • func NewPipeline(concurrency int) *Pipeline
  • func (p *Pipeline) In() chan<- interface{}
  • func (p *Pipeline) Out() <-chan interface{}
  • func (p *Pipeline) Pipe(job Job) *Pipeline

Types

type Job

type Job func(msg interface{}) interface{}

Job is the 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

func NewPipeline(concurrency int) *Pipeline

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

func (*Pipeline) Out

func (p *Pipeline) Out() <-chan interface{}

Out returns exit of the pipeline - channel with results of the last pipe

func (*Pipeline) Pipe

func (p *Pipeline) Pipe(job Job) *Pipeline

Pipe adds new pipe to pipeline with the callback for processing each message