buffer

README

This is a fork of https://github.com/globocom/go-buffer using Generics.

buffer

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/ivanvanderbyl/buffer

Examples

Size-triggered flush

package main

import (
  "time"

  "github.com/ivanvanderbyl/buffer"
)

func main() {
  buff := buffer.New(
  // call this function when the buffer needs flushing
  buffer.Fn(func(items []string) {
   println("flushing", len(items), "items")
   for _, item := range items {
    println(item)
   }
  }),
  // buffer can hold up to 5 items
  buffer.WithSize(5),
 )
 // 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")
 buff.Push("item 6") // This item will be flushed by the Closer.

 println("exiting...")
}

Interval-triggered flush

package main

import (
  "time"

  "github.com/ivanvanderbyl/buffer"
)

func main() {
  buff := buffer.New(
  // call this function when the buffer needs flushing
  buffer.Fn(func(items []string) {
   println("flushing", len(items), "items")
   for _, item := range items {
    println(item)
   }
  }),
  // buffer can hold up to 3 items
  buffer.WithSize(3),
  buffer.WithFlushInterval(time.Second),
 )
 // ensure the buffer
 defer buff.Close()

 buff.Push("item 1") // Flushed on timeout of 1 second
 buff.Push("item 2") // Flushed on timeout of 1 second
 time.Sleep(2 * time.Second)
 buff.Push("item 3")
 buff.Push("item 4")
 buff.Push("item 5")
 buff.Push("item 6") // Flushed on close

 println("exiting...")
}

Manual flush

package main

import (
 "github.com/ivanvanderbyl/buffer"
)

func main() {
 buff := buffer.New(
  buffer.Fn(func(items []string) {
   println("flushing", len(items), "items")
   for _, item := range items {
    println(item)
   }
  }),
  // buffer can hold up to 5 items
  buffer.WithSize(5),
 )
 defer buff.Close()

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

 println("done")
}

Custom Flusher

package main

import (
 "github.com/ivanvanderbyl/buffer"
)

type CustomFlusher struct{}

func (f CustomFlusher) Write(items []string) {
 println("flushing", len(items), "items")
 for _, item := range items {
  println(item)
 }
}

// Verify that CustomFlusher implements the buffer.Flusher interface
var _ buffer.Flusher[string] = (*CustomFlusher)(nil)

func main() {
 flusher := CustomFlusher{}

 buff := buffer.New[string](flusher, buffer.WithSize(5))
 defer buff.Close()

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

Documentation

Visit Pkg.go.dev for full documentation.

License

MIT License

Documentation

Index

• Variables
• type Buffer
  • func New[T any](flusher Flusher[T], opts ...Option) *Buffer[T]
  • func (buffer *Buffer[T]) Close() error
  • func (buffer *Buffer[T]) Flush() error
  • func (buffer *Buffer[T]) Push(item T) error
• type Flusher
  • func Fn[T any](fn FlusherFunc[T]) Flusher[T]
• type FlusherFunc
  • func (fn FlusherFunc[T]) Write(items []T)
• type Option
  • func WithCloseTimeout(timeout time.Duration) Option
  • func WithFlushInterval(interval time.Duration) Option
  • func WithFlushTimeout(timeout time.Duration) 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[T any] struct {
	// contains filtered or unexported fields
}

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

func New

func New[T any](flusher Flusher[T], opts ...Option) *Buffer[T]

New creates a new buffer instance with the provided options.

func (*Buffer[T]) Close

func (buffer *Buffer[T]) 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[T]) Flush

func (buffer *Buffer[T]) 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[T]) Push

func (buffer *Buffer[T]) Push(item T) 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[T any] interface {
	Write(items []T)
}

Flusher represents a destination of buffered data.

func Fn

func Fn[T any](fn FlusherFunc[T]) Flusher[T]

type FlusherFunc

type FlusherFunc[T any] func(items []T)

FlusherFunc represents a flush function.

func (FlusherFunc[T]) Write

func (fn FlusherFunc[T]) Write(items []T)

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 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
	FlushInterval time.Duration
	PushTimeout   time.Duration
	FlushTimeout  time.Duration
	CloseTimeout  time.Duration
}

Configuration options.