pipe

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"
)

func zip(r io.Reader, w io.Writer) error {
    gzw, err := gzip.NewWriterLevel(w, gzip.BestSpeed)
    if err != nil {
        return err
    }
    defer gzw.Close()
    _, err = io.Copy(gzw, r)
    return err
}

func main() {

    // pipe input
    in, err := os.Open("test.txt")
    if err != nil {
        log.Fatal(err)
    }
    defer in.Close()

    // pipe output
    out, err := os.Create("test.txt.tgz")
    if err != nil {
        log.Fatal(err)
    }
    defer out.Close()

    // create a new pipe with a io.Reader
    // Push a transformation function
    // Set output
    // Exec and get errors if any
    if err := 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"
)

func DemoRW() {

	in, err := os.Open("test.txt")
	if err != nil {
		log.Fatal(err)
	}
	defer in.Close()

	file := &rw.File{AllowSub: true}

	// Always start before use. Note that an RW after Start can be reused.
	if err := file.Start(); err != nil {
		log.Fatal(err)
	}

	// Obtain a writer
	w, err := file.NewWriter("copy.txt")
	if err != nil {
		log.Fatal(err)
	}

	// ToCloser() closes the connection at the end of the write.
	if err := pipe.New(binReader).ToCloser(w).Exec(); err != nil {
		log.Fatal(err)
	}
}

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