buffer

README

go-buffer

go-buffer represents a buffer that asynchronously flushes its contents. It is useful for applications that need to aggregate data before writing it to an external storage. A buffer is flushed manually, or automatically when it becomes full or after an interval has elapsed, whichever comes first.

Installation

go get github.com/globocom/go-buffer

Examples

Size-triggered flush

package main

import (
  "time"

  "github.com/globocom/go-buffer/v2"
)

func main() {
  buff := buffer.New(
    // buffer can hold up to 5 items
    buffer.WithSize(5),
    // call this function when the buffer needs flushing
    buffer.WithFlusher(func(items []interface{}) {
      for _, item := range items {
        println(item.(string))
      }
    }),
  )
  // ensure the buffer
  defer buff.Close()

  buff.Push("item 1")
  buff.Push("item 2")
  buff.Push("item 3")
  buff.Push("item 4")
  buff.Push("item 5")

  // block the current goroutine
  time.Sleep(3*time.Second)

  println("done")
}

Interval-triggered flush

package main

import (
  "time"

  "github.com/globocom/go-buffer/v2"
)

func main() {
  buff := buffer.New(
    // buffer can hold up to 5 items
    buffer.WithSize(5),
    // buffer will be flushed every second, regardless of
    // how many items were pushed
    buffer.WithFlushInterval(time.Second),
    // call this function when the buffer needs flushing
    buffer.WithFlusher(func(items []interface{}) {
      for _, item := range items {
        println(item.(string))
      }
    }),
  )
  defer buff.Close()

  buff.Push("item 1")
  buff.Push("item 2")
  buff.Push("item 3")

  // block the current goroutine
  time.Sleep(3*time.Second)

  println("done")
}

Manual flush

package main

import (
  "time"

  "github.com/globocom/go-buffer/v2"
)

func main() {
  buff := buffer.New(
    // buffer can hold up to 5 items
    buffer.WithSize(5),
    // call this function when the buffer needs flushing
    buffer.WithFlusher(func(items []interface{}) {
      for _, item := range items {
        println(item.(string))
      }
    }),
  )
  defer buff.Close()

  buff.Push("item 1")
  buff.Push("item 2")
  buff.Push("item 3")

  // block the current goroutine
  time.Sleep(3*time.Second)

  buff.Flush()
  println("done")
}

Documentation

Visit Pkg.go.dev for full documentation.

License

MIT License

Documentation

Index

• Variables
• type Buffer
  • func New(opts ...Option) *Buffer
  • func (buffer *Buffer) Close() error
  • func (buffer *Buffer) Flush() error
  • func (buffer *Buffer) Push(item interface{}) error
• type Flusher
• type FlusherFunc
  • func (fn FlusherFunc) Write(items []interface{})
• type Option
  • func WithCloseTimeout(timeout time.Duration) Option
  • func WithFlushInterval(interval time.Duration) Option
  • func WithFlushTimeout(timeout time.Duration) Option
  • func WithFlusher(flusher Flusher) Option
  • func WithPushTimeout(timeout time.Duration) Option
  • func WithSize(size uint) Option
• type Options

Variables

var (
	// ErrTimeout indicates an operation has timed out.
	ErrTimeout = errors.New("operation timed-out")
	// ErrClosed indicates the buffer is closed and can no longer be used.
	ErrClosed = errors.New("buffer is closed")
)

Types

type Buffer

type Buffer struct {
	io.Closer
	// contains filtered or unexported fields
}

Buffer represents a data buffer that is asynchronously flushed, either manually or automatically.

func New

func New(opts ...Option) *Buffer

New creates a new buffer instance with the provided options.

func (*Buffer) Close

func (buffer *Buffer) Close() error

Close flushes the buffer and prevents it from being further used.

It returns an ErrTimeout if if cannot be performed in a timely fashion, and an ErrClosed if the buffer has already been closed.

An ErrTimeout can either mean that a flush could not be triggered, or it can mean that a flush was triggered but it has not finished yet. In any case it is safe to call Close again.

func (*Buffer) Flush

func (buffer *Buffer) Flush() error

Flush outputs the buffer to a permanent destination.

It returns an ErrTimeout if if cannot be performed in a timely fashion, and an ErrClosed if the buffer has been closed.

func (*Buffer) Push

func (buffer *Buffer) Push(item interface{}) error

Push appends an item to the end of the buffer.

It returns an ErrTimeout if if cannot be performed in a timely fashion, and an ErrClosed if the buffer has been closed.

type Flusher

type Flusher interface {
	Write(items []interface{})
}

Flusher represents a destination of buffered data.

type FlusherFunc

type FlusherFunc func(items []interface{})

FlusherFunc represents a flush function.

func (FlusherFunc) Write

func (fn FlusherFunc) Write(items []interface{})

type Option

type Option func(*Options)

Option setter.

func WithCloseTimeout

func WithCloseTimeout(timeout time.Duration) Option

WithCloseTimeout sets how long

func WithFlushInterval

func WithFlushInterval(interval time.Duration) Option

WithFlushInterval sets the interval between automatic flushes.

func WithFlushTimeout

func WithFlushTimeout(timeout time.Duration) Option

WithFlushTimeout sets how long a manual flush should wait before giving up.

func WithFlusher

func WithFlusher(flusher Flusher) Option

WithFlusher sets the flusher that should be used to write out the buffer.

func WithPushTimeout

func WithPushTimeout(timeout time.Duration) Option

WithPushTimeout sets how long a push should wait before giving up.

func WithSize

func WithSize(size uint) Option

WithSize sets the size of the buffer.

type Options

type Options struct {
	Size          uint
	Flusher       Flusher
	FlushInterval time.Duration
	PushTimeout   time.Duration
	FlushTimeout  time.Duration
	CloseTimeout  time.Duration
}

Configuration options.