go-resiliency: github.com/eapache/go-resiliency/batcher Index | Examples | Files

package batcher

import "github.com/eapache/go-resiliency/batcher"

Package batcher implements the batching resiliency pattern for Go.

Index

Examples

Package Files

batcher.go

type Batcher Uses

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

Batcher implements the batching resiliency pattern

Code:

package main

import (
    "errors"
    "sync"
    "sync/atomic"
    "testing"
    "time"
)

var errSomeError = errors.New("errSomeError")

func returnsError(params []interface{}) error {
    return errSomeError
}

func returnsSuccess(params []interface{}) error {
    return nil
}

func TestBatcherSuccess(t *testing.T) {
    b := New(10*time.Millisecond, returnsSuccess)

    wg := &sync.WaitGroup{}
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func() {
            if err := b.Run(nil); err != nil {
                t.Error(err)
            }
            wg.Done()
        }()
    }
    wg.Wait()

    b = New(0, returnsSuccess)
    for i := 0; i < 10; i++ {
        if err := b.Run(nil); err != nil {
            t.Error(err)
        }
    }
}

func TestBatcherError(t *testing.T) {
    b := New(10*time.Millisecond, returnsError)

    wg := &sync.WaitGroup{}
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func() {
            if err := b.Run(nil); err != errSomeError {
                t.Error(err)
            }
            wg.Done()
        }()
    }
    wg.Wait()
}

func TestBatcherPrefilter(t *testing.T) {
    b := New(1*time.Millisecond, returnsSuccess)

    b.Prefilter(func(param interface{}) error {
        if param == nil {
            return errSomeError
        }
        return nil
    })

    if err := b.Run(nil); err != errSomeError {
        t.Error(err)
    }

    if err := b.Run(1); err != nil {
        t.Error(err)
    }
}

func TestBatcherMultipleBatches(t *testing.T) {
    var iters uint32

    b := New(10*time.Millisecond, func(params []interface{}) error {
        atomic.AddUint32(&iters, 1)
        return nil
    })

    wg := &sync.WaitGroup{}

    for group := 0; group < 5; group++ {
        for i := 0; i < 10; i++ {
            wg.Add(1)
            go func() {
                if err := b.Run(nil); err != nil {
                    t.Error(err)
                }
                wg.Done()
            }()
        }
        time.Sleep(15 * time.Millisecond)
    }

    wg.Wait()

    if iters != 5 {
        t.Error("Wrong number of iters:", iters)
    }
}

func main() {
    b := New(10*time.Millisecond, func(params []interface{}) error {
        // do something with the batch of parameters
        return nil
    })

    b.Prefilter(func(param interface{}) error {
        // do some sort of sanity check on the parameter, and return an error if it fails
        return nil
    })

    for i := 0; i < 10; i++ {
        go b.Run(i)
    }
}

func New Uses

func New(timeout time.Duration, doWork func([]interface{}) error) *Batcher

New constructs a new batcher that will batch all calls to Run that occur within `timeout` time before calling doWork just once for the entire batch. The doWork function must be safe to run concurrently with itself as this may occur, especially when the timeout is small.

func (*Batcher) Prefilter Uses

func (b *Batcher) Prefilter(filter func(interface{}) error)

Prefilter specifies an optional function that can be used to run initial checks on parameters passed to Run before being added to the batch. If the prefilter returns a non-nil error, that error is returned immediately from Run and the batcher is not invoked. A prefilter cannot safely be specified for a batcher if Run has already been invoked. The filter function specified must be concurrency-safe.

func (*Batcher) Run Uses

func (b *Batcher) Run(param interface{}) error

Run runs the work function with the given parameter, possibly including it in a batch with other calls to Run that occur within the specified timeout. It is safe to call Run concurrently on the same batcher.

Package batcher imports 2 packages (graph). Updated 2020-03-14. Refresh now. Tools for package owners.