tpack

README

tpack

Pack a Go workflow/function into a Unix-style pipeline command.

Wiki
In Unix-like computer operating systems, a pipeline is a mechanism for inter-process communication using message passing. A pipeline is a set of processes chained together by their standard streams, so that the output text of each process (stdout) is passed directly as input (stdin) to the next one.

Use tpack to write Go applications that act as pipeline commands. Employ channels, goroutines, regular expressions and more to build powerful concurrent workflows.

Usage

See the ETL workflow in the examples folder.

package main

import "github.com/reugn/tpack"

func main() {
	tpack.NewPackerStd(tpack.NewProcessor(
		doETL,
	)).Execute()
}

Test command

cat input.txt | go run *.go 2>/dev/null | wc -l

License

Licensed under the MIT License.

Documentation

Overview

Package tpack provides tools to package Go workflows as Unix-style pipeline commands.

Index

• type Packer
  • func NewPacker(in io.Reader, out, err io.Writer, processor Processor, opts ...PackerOpt) *Packer
  • func NewPackerStd(processor Processor, opts ...PackerOpt) *Packer
  • func NewPackerStdOut(in io.Reader, processor Processor, opts ...PackerOpt) *Packer
  • func (p *Packer) Execute()
• type PackerOpt
  • func WithErrWriteHandler(f func(error)) PackerOpt
• type ProcOpt
  • func Parallel(p int) ProcOpt
• type Processor
  • func NewProcessor[T data](transform func(T) ([]T, error), opts ...ProcOpt) Processor

Types

type Packer

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

Packer facilitates writing programs that manipulate text streams, a fundamental concept in the Unix philosophy. It enables seamless data transfer between processes, allowing the output of one process to be used as input for another.

func NewPacker

func NewPacker(in io.Reader, out, err io.Writer, processor Processor,
	opts ...PackerOpt) *Packer

NewPacker returns a new Packer using custom communication channels.

func NewPackerStd

func NewPackerStd(processor Processor, opts ...PackerOpt) *Packer

NewPackerStd returns a new Packer that uses the standard input, output, and error streams as communication channels.

func NewPackerStdOut

func NewPackerStdOut(in io.Reader, processor Processor,
	opts ...PackerOpt) *Packer

NewPackerStdOut returns a new Packer that uses the standard output and error streams as output channels and the provided io.Reader as the input communication channel.

func (*Packer) Execute

func (p *Packer) Execute()

Execute starts processing data stream.

type PackerOpt

type PackerOpt func(*packerOpts)

PackerOpt represents a customization option to configure a packer.

func WithErrWriteHandler

func WithErrWriteHandler(f func(error)) PackerOpt

WithErrWriteHandler configures a custom error handler for writing to err. The default error handler will panic on an error.

type ProcOpt

type ProcOpt func(*procOpts)

ProcOpt represents a customization option to configure a processor.

func Parallel

func Parallel(p int) ProcOpt

Parallel configures the processor parallelism. The specified value is required to be greater than zero. Otherwise, it is ignored. The default parallelism is 1.

Parallel workflows can be useful when the processing order is not important - for further counting or any other type of aggregation command.

type Processor

type Processor interface {
	// InChan returns the input communication channel.
	InChan() chan []byte

	// OutChan returns the output communication channel.
	OutChan() chan []byte

	// ErrChan returns the error output communication channel.
	ErrChan() chan error
}

Processor represents a generic data stream processor.

func NewProcessor

func NewProcessor[T data](transform func(T) ([]T, error), opts ...ProcOpt) Processor

NewProcessor returns a new Processor with the specified transformation function. The transformation function type can be either string or slice of bytes. Additional options can be provided to customize the processor, e.g.

tpack.Parallel(2)