buffer/lib.go

// Package buffer provides pooled Buffer IO for Go programs.
//
// It implements a reader similar to bufio.Reader. The underlying memory buffers can be used from a synchronized
// pool.
package buffer

import (
	"errors"
	"io"
)

// Pool defines the interface for the used buffer pool. The buffered reader can be used either with the built-in
// default pool, noop pool, or with a custom pool implementation.
type Pool interface {

	// Get should return a non-zero length byte slice. In case it returns a zero length byte slice, or an
	// explicit error, the read operations will fail.
	//
	// It is OK but not recommended to return varying sizes of byte slices.
	Get() ([]byte, error)

	// The reader always puts back the byte slices taken by Get, using Put.
	Put([]byte)
}

// Options provides options for the Reader.
type Options struct {

	// Pool defines the buffer pool to be used. It defaults to the pool created by DefaultPool().
	Pool Pool
}

// ContentFunc wraps a function to implement the io.WriterTo interface. Function implementations should be
// reader to be executed in goroutines other than what they were created in.
type ContentFunc func(io.Writer) (int64, error)

// Reader wraps an underlying io.Reader or io.WriterTo, and provides buffered io, via its methods. Initialize it
// via BufferedReader or BufferedContent.
//
// It reads from the underlying source until the first error, but only returns an error when the buffer is
// empty. Once the underlying reader returned an error, it doesn't attempt to read from it anymore.
//
// The reader does not support concurrent access.
type Reader struct {
	reader *reader
}

var (
	// ErrZeroAllocation is returned when the used pool returned a zero length byte slice.
	ErrZeroAllocation = errors.New("zero allocation")

	// ErrContentAbort is returned to the writer process in case of buffered content, when the reader
	// experienced an error.
	ErrContentAbort = errors.New("content pipe aborted")
)

// DefultPool initializes a synchronized pool that stores and returns byte slices of allocSize length. It can be
// used with multiple readers concurrently.
func DefaultPool(allocSize int) Pool {
	return newPool(allocSize)
}

// NoPool returns a noop pool.
func NoPool(allocSize int) Pool {
	return noPool{allocSize: allocSize}
}

// WriteTo implements the WriterTo interface.
func (f ContentFunc) WriteTo(w io.Writer) (int64, error) {
	return f(w)
}

// BufferedReader creates a buffered reader using the input reader as the underlying source.
func BufferedReader(in io.Reader, o Options) Reader {
	if in == nil {
		return Reader{}
	}

	if o.Pool == nil {
		o.Pool = DefaultPool(1 << 12)
	}

	return Reader{reader: &reader{options: o, in: in}}
}

// BufferedContent creates a buffered reader using the input content (io.WriterTo) as the underlying source.
//
// It is similar to an io.Pipe, but with dynamic and pooled buffering internally. The individual Write calls are
// blocked until the reading side requests more data.
//
// The provided WriterTo instances need to be safe to call in goroutines other than they were created in. The
// writer function returns with nil error, it will be interpreted as EOF on the reader side. When the reader
// side experiences an error, and the writer still has content to be written, the passed in io.Writer will
// return an ErrContentAbort error.
func BufferedContent(c io.WriterTo, o Options) Reader {
	if c == nil {
		return Reader{}
	}

	if o.Pool == nil {
		o.Pool = DefaultPool(1 << 12)
	}

	return Reader{reader: &reader{options: o, in: mkcontent(c)}}
}

// Read reads max len(p) copied to p and returns how many bytes were read.
//
// It only returns an error when the buffer is empty. It only returns an error when the Pool.Get fails or the
// underlying reader returns an error.
//
// It may return zero read length and nil error, but only if the underlying reader did so.
func (r Reader) Read(p []byte) (int, error) {
	if r.reader == nil {
		return 0, io.EOF
	}

	return r.reader.read(p)
}

// ReadBytes reads until the first occurence of delimiter in the input, within max length.
//
// It returns the bytes, true and nil error, when the delimiter was found within max. In this case it consumes
// the buffer until and including the delimiter. If the underlying reader returned meanwhile a non-nil error,
// including EOF, it will be returned on subsequent reads, but only after the internal buffer was consumed.
//
// It returns zero length bytes, false and nil error, when the delimiter was not found within max, and the
// underlying reader didn't return an error.
//
// It returns max or less bytes, false and nil error, if the delimiter was not found within max, there are
// buffered bytes, the underlying reader returned an error. It consumes the returned bytes from the buffer. The
// delimiter may still be found in the remaining buffered bytes during subsequent calls.
//
// It returns zero length bytes, false and non-nil error, if the buffer is empty and the underlying reader
// previously returned an error.
func (r Reader) ReadBytes(delimiter []byte, max int) ([]byte, bool, error) {
	if r.reader == nil {
		return nil, false, io.EOF
	}

	return r.reader.readBytes(delimiter, max)
}

// ReadUTF8 returns at most max UTF8 characters read from the underlying reader.
//
// When valid UTF8 characters were found, it returns the characters, the number of bytes consumed, and nil
// error.
//
// It only returns an error, including EOF, when the underlying buffer is empty.
//
// It may return zero length characters and nil error, if the underlying reader returned zero length read with
// nil error.
//
// It supports recovery of UTF8 streams by skipping the invalid characters. In such cases, first it returns the
// valid characters with the number of bytes consumed and nil error, then in the subsequent call, it returns
// zero characters, 1 and nil error.
func (r Reader) ReadUTF8(max int) ([]rune, int, error) {
	if r.reader == nil {
		return nil, 0, io.EOF
	}

	return r.reader.readUTF8(max)
}

// Peek returns at most max read bytes and keeps them buffered.
//
// If the underlying reader returned an error, it returns an error when the underlying buffer was fully consumed
// by Read, ReadBytes, ReadUTF8 or WriteTo.
//
// The returned byte slice is not a copy of the buffered bytes, and therefore should not be modified.
func (r Reader) Peek(max int) ([]byte, error) {
	if r.reader == nil {
		return nil, io.EOF
	}

	return r.reader.peek(max)
}

// Buffered returns the buffered bytes.
//
// The returned byte slice is not a copy of the buffered bytes, and therefore should not be modified.
func (r Reader) Buffered() []byte {
	if r.reader == nil {
		return nil
	}

	return r.reader.buffered()
}

// WriteTo implements the io.WriterTo interface. It copies all the data from the underlying reader to the
// provided writer using a buffer from the pool.
//
// It is important that the provided writer must not modify the slice data, as defined in the io.Writer
// interface documentation.
func (r Reader) WriteTo(w io.Writer) (int64, error) {
	if r.reader == nil {
		return 0, nil
	}

	return r.reader.writeTo(w)
}