cookiemonster

README

CookieMonster

Purpose

CookieMonster is a flexible worker pool that can be adapted to fit (almost) any usecase.

This is accomplished with the use of Jars and Cookies. These are simply interfaces that represent a work provider and the information needed to process that work, respectively.

Usage

1. Add CookieMonster to your project:

   go get -u github.com/cookiejars/cookie-monster

2. Setup your Jar, by either:

   a) Checking if there's already an implementation that fits your usecase.

   b) Implementing your own Jar. For this you'll need to create implementations that fit the Jar and Cookie interfaces.

   // Represents a work provider
   type Jar interface {
       Retrieve() ([]Cookie, error) // generate a unit of work to distribute amongst the various workers
       Retire(Cookie) error         // mark the work as done (e.g., delete a message from a queue after it's been processed)
   }
   
   // Represents a unit of work
   type Cookie interface {
       ID() string                  // work identifier
       Content() interface{}        // data needed to process the work
       Metadata() map[string]string // optional map of metadata related to the work
   }
   

   Please consider sharing your implementation as an open-source project and help this baby grow (check Contributing).

3. Initialize the pool:

   digester := cookiemonster.NewDigester(jar)
   

   Optionally, you can also pass a list of options to tweak how the digester works internally.

   List of options:

   Function	Description	Default behaviour (if not called)
   SetWorkers	Define the number of goroutines processing work concurrently.	Sets to runtime.NumCPU().
   SetBackoff	Define how the digester behaves in the interval of processing work. You can use the ConstantBackoff or ExponentialBackoff implementations provided or use your own implementation of the Backoff interface.	Sets to ConstantBackoff of 10 seconds.
   SetInfoLog	Define info logger where internal information is written to.	No info logs.
   SetErrorLog	Define error logger where errors are written to.	No error logs.
   SetStopSignals	Define a list of os.Signals for the digester to wait for to call Stop() automatically, once you call Start().	Start() will exit immediately, leaving the pool working in the background. It will be up to you to define how to wait for the work to be complete and call digester.Stop() to stop the pool.

4. Start the pool:

   digester.Start(digestFn)
   

   Here you have to pass the function used to process the work, in the form of: func(cookie cookiemonster.Cookie) error.

5. Stop the pool (optional):

   digester.Stop()
   

   If you didn't pass any signals, you'll need to explicitly call the Stop function for the pool to gracefully shutdown.

   On the other hand, if you did pass them, this will be automatically called once any of them is triggered.

Examples

Check the examples folder.

Contributing

This project aims to be generic and fit as much cases as possible. This will only be possible if you share your specific usecase to help identify where the project is still lacking.

To ease discovery, I'd love to have github.com/cookiejars as the main place to go to find the existent implementations, so if you wish to contribute feel free to open an issue or DM me on the Gophers' Slack.

Documentation

Index

• type Backoff
  • func ConstantBackoff(amount time.Duration) Backoff
  • func ExponentialBackoff(n int, initialAmount time.Duration) Backoff
• type Cookie
• type DigestFn
• type Digester
  • func NewDigester(jar Jar, options ...DigesterOptionFunc) Digester
  • func NewDigesterWithPlugin(jarPath string, options ...DigesterOptionFunc) (Digester, error)
• type DigesterOptionFunc
  • func SetBackoff(backoff Backoff) DigesterOptionFunc
  • func SetErrorLog(logger Logger) DigesterOptionFunc
  • func SetInfoLog(logger Logger) DigesterOptionFunc
  • func SetStopSignals(signals ...os.Signal) DigesterOptionFunc
  • func SetWorkers(workers int) DigesterOptionFunc
• type Jar
• type Logger

Types

type Backoff

type Backoff interface {
	Next()
	Current() time.Duration
	Reset()
}

func ConstantBackoff

func ConstantBackoff(amount time.Duration) Backoff

func ExponentialBackoff

func ExponentialBackoff(n int, initialAmount time.Duration) Backoff

type Cookie

type Cookie interface {
	ID() string
	Content() interface{}
	Metadata() map[string]string
}

type DigestFn

type DigestFn func(Cookie) error

type Digester

type Digester interface {
	Start(fn DigestFn) error
	Stop()
}

func NewDigester

func NewDigester(jar Jar, options ...DigesterOptionFunc) Digester

func NewDigesterWithPlugin

func NewDigesterWithPlugin(jarPath string, options ...DigesterOptionFunc) (Digester, error)

type DigesterOptionFunc

type DigesterOptionFunc func(*digester)

func SetBackoff

func SetBackoff(backoff Backoff) DigesterOptionFunc

func SetErrorLog

func SetErrorLog(logger Logger) DigesterOptionFunc

func SetInfoLog

func SetInfoLog(logger Logger) DigesterOptionFunc

func SetStopSignals

func SetStopSignals(signals ...os.Signal) DigesterOptionFunc

func SetWorkers

func SetWorkers(workers int) DigesterOptionFunc

type Jar

type Jar interface {
	Retrieve() ([]Cookie, error)
	Retire(Cookie) error
}

type Logger

type Logger interface {
	Printf(format string, args ...interface{})
}