batchio

README

zombiezen.com/go/batchio Package

The Go package batchio provides mechanisms for converting a stream of bytes into batches of approximately equal time and space. This provides a reasonable balance between throughput and latency.

License

BSD 3-Clause. This package was spun out from github.com/yourbase/commons.

Documentation

Overview

Package batchio provides mechanisms for converting a stream of bytes into batches of approximately equal time and space. This provides a reasonable balance between throughput and latency.

Index

• type Reader
  • func NewReader(r io.ReadCloser, size int, timeAfterFirstByte time.Duration) *Reader
  • func (r *Reader) Finish() ([]byte, error)
  • func (r *Reader) Next(ctx context.Context) ([]byte, error)
• type Writer
  • func NewWriter(w io.Writer, size int, timeAfterFirstByte time.Duration) *Writer
  • func (w *Writer) Flush() error
  • func (w *Writer) Write(p []byte) (n int, err error)

Examples

• Reader

Types

type Reader

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

A Reader buffers an io.Reader to produce a sequence of batches.

Example

package main

import (
	"context"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"os"
	"strings"
	"time"

	"zombiezen.com/go/batchio"
)

func main() {
	ctx := context.Background()

	// The stream can be any io.ReadCloser that supports calling Close
	// concurrently with Read. Examples from the standard library include
	// *os.File, net.Conn, *io.PipeReader, and net/http.Request.Body.
	stream := ioutil.NopCloser(strings.NewReader("Hello, World!"))

	// Set parameters for your batches.
	const maxBatchSize = 5
	const timeAfterFirstByte = 10 * time.Second
	reader := batchio.NewReader(stream, maxBatchSize, timeAfterFirstByte)

	// Always call Finish to close the stream and read any buffered data.
	defer func() {
		last, err := reader.Finish()
		if len(last) > 0 {
			fmt.Printf("%s\n", last)
		}
		if err != nil {
			fmt.Fprintln(os.Stderr, "Finish error:", err)
			return
		}
	}()

	// Loop until stream encounters an error.
	for {
		batch, err := reader.Next(ctx)
		if err != nil {
			if !errors.Is(err, io.EOF) {
				fmt.Fprintln(os.Stderr, "Error:", err)
			}
			break
		}
		fmt.Printf("%s\n", batch)
	}

}

Output:

Hello
, Wor
ld!

func NewReader

func NewReader(r io.ReadCloser, size int, timeAfterFirstByte time.Duration) *Reader

NewReader returns a new Reader that reads batches from r. The batches will be no larger than the given size and will wait at most the given time after the first byte before returning.

It must be safe to call r.Close concurrently with r.Read.

func (*Reader) Finish

func (r *Reader) Finish() ([]byte, error)

Finish closes the underlying reader and returns a final batch if a Read was pending. After the first call to Finish, it returns an error.

func (*Reader) Next

func (r *Reader) Next(ctx context.Context) ([]byte, error)

Next reads the next batch from c's underlying reader. Next reads until its buffer is full, the duration after the first byte has elapsed, its underlying reader returns an error, or the Context is Done, whichever comes first. The returned batch is valid until the next call to Next.

Next will return either a batch or an error. Once the underlying reader has returned an error, the Next will return the same error on subsequent calls.

type Writer

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

A Writer is a buffered io.Writer that writes batches to an underlying io.Writer object. If an error occurs writing to a Writer, no more data will be accepted and all subsequent writes, and Flush, will return the error. After all data has been written, the client should call the Writer.Flush method to guarantee all data has been forwarded to the underlying io.Writer object.

func NewWriter

func NewWriter(w io.Writer, size int, timeAfterFirstByte time.Duration) *Writer

NewWriter returns a new Writer that writes batches to w. The batches will be no larger than the given size and will wait at most the given time after the first byte in a batch before writing the whole batch.

func (*Writer) Flush

func (w *Writer) Flush() error

Flush writes any buffered data to the underlying io.Writer.

func (*Writer) Write

func (w *Writer) Write(p []byte) (n int, err error)

Write writes the contents of p into the buffer. It returns the number of bytes written. If n < len(p), it also returns an error explaining why the write is short.