parallel_chunked_flow

README

parallel-chunked-flow

Package parallel chunked flow provides the ability to craete parallel pipeline for sequential data.

Benchmark

Here is result of benchmark with parsing JSON string:

$ go test -bench=.
goos: darwin
goarch: amd64
pkg: github.com/cfsghost/parallel-chunked-flow
BenchmarkBasic/Small-16                      	  721980	      1516 ns/op
BenchmarkBasic/Large-16                      	    9712	    110151 ns/op
BenchmarkChunkedFlowWithLowChunkCount/Small-16         	 2427859	       441 ns/op
BenchmarkChunkedFlowWithLowChunkCount/Large-16         	   52950	     31015 ns/op
BenchmarkChunkedFlowWithHighChunkCount/Small-16        	 2539164	       449 ns/op
BenchmarkChunkedFlowWithHighChunkCount/Large-16        	   54291	     26143 ns/op
PASS
ok  	github.com/cfsghost/parallel-chunked-flow	10.282s

License

Licensed under the MIT License

Authors

Copyright(c) 2020 Fred Chien cfsghost@gmail.com

Documentation

Index

• type Chunk
  • func NewChunk(size int) *Chunk
  • func (chunk *Chunk) Initialize()
  • func (chunk *Chunk) Output() chan interface{}
• type Options
  • func NewOptions() *Options
• type ParallelChunkedFlow
  • func NewParallelChunkedFlow(options *Options) *ParallelChunkedFlow
  • func (pcf *ParallelChunkedFlow) Close()
  • func (pcf *ParallelChunkedFlow) Output() chan interface{}
  • func (pcf *ParallelChunkedFlow) Push(data interface{}) error

Types

type Chunk

type Chunk struct {
	Handler func(interface{}, func(interface{}))
	// contains filtered or unexported fields
}

func NewChunk

func NewChunk(size int) *Chunk

func (*Chunk) Initialize

func (chunk *Chunk) Initialize()

func (*Chunk) Output

func (chunk *Chunk) Output() chan interface{}

type Options

type Options struct {
	BufferSize int
	ChunkSize  int
	ChunkCount int
	Handler    func(interface{}, func(interface{}))
}

Options represent all of the available options when creating a parallel chunked flow.

func NewOptions

func NewOptions() *Options

NewOptions creates a Options object.

type ParallelChunkedFlow

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

func NewParallelChunkedFlow

func NewParallelChunkedFlow(options *Options) *ParallelChunkedFlow

NewParallelChunckedFlow creates a new parallel chunked flow

func (*ParallelChunkedFlow) Close

func (pcf *ParallelChunkedFlow) Close()

Close all goroutines

func (*ParallelChunkedFlow) Output

func (pcf *ParallelChunkedFlow) Output() chan interface{}

Output will return a channel for receive proccessed data from flow

func (*ParallelChunkedFlow) Push

func (pcf *ParallelChunkedFlow) Push(data interface{}) error

Push will put data to the flow