stream

Documentation

Overview

Package stream implements a Reader and Writer for simple streams of either length-delimited byte records (using shipshape/util/delimited) or newline-delimited JSON values.

Index

• type Reader
  • func NewReader(r io.Reader, decodeJSON bool) Reader
  • func Transform(rd Reader, f func([]byte) ([]byte, error)) Reader
• type Writer
  • func NewWriter(w io.Writer, lineFormat bool) Writer

Types

type Reader

type Reader interface {
	// Next returns the next record from the input, or io.EOF if there are no
	// more records available.
	//
	// The slice returned is valid only until a subsequent call to Next.
	Next() ([]byte, error)
}

Reader provides sequential access to a stream of RPC values.

Usage:

rd := stream.NewReader(r, false)
for {
  rec, err := rd.Next()
  if err == io.EOF {
    break
  } else if err != nil {
    log.Fatal(err)
  }
  doStuffWith(rec)
}

func NewReader

func NewReader(r io.Reader, decodeJSON bool) Reader

NewReader constructs a new stream Reader for the records in r. If decodeJSON is true, then the Reader is considered a stream of JSON arbitrary values; otherwise the Reader is considered a length-delimited stream of byte records.

func Transform

func Transform(rd Reader, f func([]byte) ([]byte, error)) Reader

Transform returns a new Reader that applies the given func to each stream value before returning it. Errors are passed through.

type Writer

type Writer interface {
	// Put writes the specified record to the Writer's stream.
	Put(data []byte) error
}

A Writer outputs delimited records to an io.Writer.

Basic usage:

wr := stream.NewWriter(w, false)
for record := range records {
   if err := wr.Put(record); err != nil {
     log.Fatal(err)
   }
}

func NewWriter

func NewWriter(w io.Writer, lineFormat bool) Writer

NewWriter constructs a new stream Writer that writes records to w. If lineFormat is true, each record will be followed by a newline; otherwise each record is prefixed by a varint representing its length in bytes.