pipe

A simple stream processing library that works like Unix pipes. This library is fully asynchronous. Create a Pipe from a Reader, add some transformation functions and get the result writed to a Writer.

To install :

$ go get github.com/hyperboloide/pipe

Then add the following import :

import"github.com/hyperboloide/pipe"

Example

Bellow is a very basic example that:

1. Open a file
2. Compress it
3. Save it

package main
import (
"compress/gzip""github.com/hyperboloide/pipe""io""log""os"
)
funczip(r io.Reader, w io.Writer) error {
gzw, err:=gzip.NewWriterLevel(w, gzip.BestSpeed)
iferr!=nil {
returnerr
}
defergzw.Close()
_, err=io.Copy(gzw, r)
returnerr
}
funcmain() {
// pipe inputin, err:=os.Open("test.txt")
iferr!=nil {
log.Fatal(err)
}
deferin.Close()
// pipe outputout, err:=os.Create("test.txt.tgz")
iferr!=nil {
log.Fatal(err)
}
deferout.Close()
// create a new pipe with a io.Reader// Push a transformation function// Set output// Exec and get errors if anyiferr:=pipe.New(in).Push(zip).To(out).Exec(); err!=nil {
log.Fatal(err)
}
}

Readers and Writers

Pipe also provides a set of Reader/Writer to read from and write to.

• S3
• Google Cloud Storage
• File

Here is an example:

import (
"github.com/hyperboloide/pipe""github.com/hyperboloide/pipe/rw""log""os"
)
funcDemoRW() {
in, err:=os.Open("test.txt")
iferr!=nil {
log.Fatal(err)
}
deferin.Close()
file:=&rw.File{AllowSub: true}
// Always start before use. Note that an RW after Start can be reused.iferr:=file.Start(); err!=nil {
log.Fatal(err)
}
// Obtain a writerw, err:=file.NewWriter("copy.txt")
iferr!=nil {
log.Fatal(err)
}
// ToCloser() closes the connection at the end of the write.iferr:=pipe.New(binReader).ToCloser(w).Exec(); err!=nil {
log.Fatal(err)
}
}

It's also easy to create your own, just implement the ReadWriteDeleter interface.