machine

README

Machine

import "github.com/autom8ter/machine"

m := machine.New(context.Background(),
	// functions are added to a FIFO channel that will block when active routines == max routines. 
	machine.WithMaxRoutines(10),
        // every function executed by machine.Go will recover from panics
	machine.WithMiddlewares(machine.PanicRecover()),
	// WithValue passes the value to the root context of the machine- it is available in the context of all child machine's & all Routine's
        machine.WithValue("testing", true),
        // WithTimeout cancels the machine's context after the given timeout
        machine.WithTimeout(30 *time.Second)
)
defer m.Close()

channelName := "acme.com"

// start a goroutine that subscribes to all messages sent to the target channel for 5 seconds
m.Go(func(routine machine.Routine) {
		routine.Subscribe(channelName, func(obj interface{}) {
			fmt.Printf("%v | subscription msg received! channel = %v msg = %v stats = %s\n",
				routine.PID(), channelName, obj, m.Stats().String())
		})
	},
	machine.GoWithTags("subscribe"),
	machine.GoWithTimeout(5*time.Second),
)

// start another goroutine that publishes to the target channel every second for 5 seconds
m.Go(func(routine machine.Routine) {
		fmt.Printf("%v | streaming msg to channel = %v stats = %s\n", routine.PID(), channelName, routine.Machine().Stats().String())
		// publish message to channel
		routine.Publish(channelName, "hey there bud!")
	},
	machine.GoWithTags("publish"),
	machine.GoWithTimeout(5*time.Second),
	machine.GoWithMiddlewares(
		// run every second until context cancels
		machine.Cron(time.NewTicker(1*time.Second)),
	),
)

m.Wait()

Machine is a zero dependency library for highly concurrent Go applications. It is inspired by errgroup.Group with extra bells & whistles:

• throttled goroutines

• self-cancellable goroutines with Context

• global-cancellable goroutines with context (see Cancel)

• goroutines have IDs and optional tags for easy debugging (see Stats)

• publish/subscribe to channels for broadcasting messages to active goroutines

• middlewares for wrapping/decorating functions

• "sub" machines for creating a dependency tree between groups of goroutines

• greater than 80% test coverage

• goroutine leak prevention

Use Cases

Machine is meant to be completely agnostic and dependency free- its use cases are expected to be emergent. Really, it can be used anywhere goroutines are used.

Highly concurrent and/or asynchronous applications include:

• gRPC streaming servers

• websocket servers

• pubsub servers

• reverse proxies

• cron jobs

• custom database/cache

• ETL pipelines

• log sink

• filesystem walker

• code generation

Examples

All examples are < 500 lines of code(excluding code generation)

• gRPC Bidirectional Chat Stream Server
• TCP Reverse Proxy
• Concurrent Cron Job Server

Documentation

Index

• Constants
• type Cache
• type Func
• type GoOpt
  • func GoWithDeadline(deadline time.Time) GoOpt
  • func GoWithMiddlewares(middlewares ...Middleware) GoOpt
  • func GoWithPID(id string) GoOpt
  • func GoWithTags(tags ...string) GoOpt
  • func GoWithTimeout(to time.Duration) GoOpt
  • func GoWithValues(key, val interface{}) GoOpt
• type Machine
  • func New(ctx context.Context, options ...Opt) *Machine
  • func (p *Machine) Active() int
  • func (m *Machine) Cache() Cache
  • func (p *Machine) Cancel()
  • func (m *Machine) CancelRoutine(id string)
  • func (m *Machine) Close()
  • func (m *Machine) Go(fn Func, opts ...GoOpt)
  • func (m *Machine) HasRoutine(id string) bool
  • func (m *Machine) ID() string
  • func (m *Machine) Parent() *Machine
  • func (m *Machine) Stats() *Stats
  • func (m *Machine) Sub(opts ...Opt) *Machine
  • func (p *Machine) Tags() []string
  • func (p *Machine) Total() int
  • func (m *Machine) Wait()
• type Middleware
  • func After(afterFunc func(routine Routine)) Middleware
  • func Before(beforeFunc func(routine Routine)) Middleware
  • func Cron(ticker *time.Ticker) Middleware
  • func Decider(deciderFunc func(routine Routine) bool) Middleware
  • func PanicRecover() Middleware
  • func While(deciderFunc func(routine Routine) bool) Middleware
• type Opt
  • func WithCache(cache Cache) Opt
  • func WithChildren(children ...*Machine) Opt
  • func WithDeadline(deadline time.Time) Opt
  • func WithID(id string) Opt
  • func WithMaxRoutines(max int) Opt
  • func WithMiddlewares(middlewares ...Middleware) Opt
  • func WithPubSub(pubsub PubSub) Opt
  • func WithTags(tags ...string) Opt
  • func WithTimeout(to time.Duration) Opt
  • func WithValue(key, val interface{}) Opt
• type PubSub
• type Routine
• type RoutineStats
• type Stats
  • func (s Stats) String() string

Examples

• New

Constants

const DefaultMaxRoutines = 1000

Types

type Cache

type Cache interface {
	// Get gets the value for the given key in the given namespace.
	Get(namespace string, key interface{}) (interface{}, bool)
	// Set sets a value for the given key in the given namespace with an expiration duration.
	// If the duration is 0 or less, it will be stored forever.
	Set(namespace string, key interface{}, value interface{}, duration time.Duration)
	// Range calls f sequentially for each key and value present within the given namespace.
	// If f returns false, range stops the iteration.
	Range(namespace string, f func(key, value interface{}) bool)
	// Delete deletes the key and its value from the given namespace.
	Delete(namespace string, key interface{})
	// Len returns total kv pairs within namespace
	Len(namespace string) int
	// Close closes the Cache and frees up resources.
	Close()
}

Cache is a concurrency safe cache that stores arbitrary, namespaced data with optional TTL.

type Func

type Func func(routine Routine)

Func is the function passed into machine.Go. The Routine is passed into this function at runtime.

type GoOpt

type GoOpt func(o *goOpts)

GoOpt is a function that configures GoOpts

func GoWithDeadline

func GoWithDeadline(deadline time.Time) GoOpt

GoWithDeadline is a GoOpt that creates the Routine's context with the given deadline.

func GoWithMiddlewares

func GoWithMiddlewares(middlewares ...Middleware) GoOpt

GoWithMiddlewares wraps the gived function with the input middlewares.

func GoWithPID

func GoWithPID(id string) GoOpt

GoWithPID is a GoOpt that sets/overrides the process ID of the Routine. A random id is assigned if this option is not used.

func GoWithTags

func GoWithTags(tags ...string) GoOpt

GoWithTags is a GoOpt that adds an array of strings as "tags" to the Routine.

func GoWithTimeout

func GoWithTimeout(to time.Duration) GoOpt

GoWithTimeout is a GoOpt that creates the Routine's context with the given timeout value

func GoWithValues

func GoWithValues(key, val interface{}) GoOpt

GoWithValues adds the k/v to the routine's root context. It can be retrieved with routine.Context().Value()

type Machine

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

Machine is a zero dependency runtime for managed goroutines. It is inspired by errgroup.Group with extra bells & whistles:

func New

func New(ctx context.Context, options ...Opt) *Machine

New Creates a new machine instance with the given root context & options

Example

package main

import (
	"context"
	"fmt"
	"github.com/autom8ter/machine"
	"time"
)

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	m := machine.New(ctx,
		machine.WithMaxRoutines(10),
		machine.WithMiddlewares(machine.PanicRecover()),
	)
	defer m.Close()

	channelName := "acme.com"

	// start a goroutine that subscribes to all messages sent to the target channel for 5 seconds
	m.Go(func(routine machine.Routine) {
		routine.Subscribe(channelName, func(obj interface{}) {
			fmt.Printf("%v | subscription msg received! channel = %v msg = %v stats = %s\n", routine.PID(), channelName, obj, m.Stats().String())
		})
	}, machine.GoWithTags("subscribe"),
		machine.GoWithTimeout(5*time.Second),
	)

	// start another goroutine that publishes to the target channel every second for 5 seconds
	m.Go(func(routine machine.Routine) {
		fmt.Printf("%v | streaming msg to channel = %v stats = %s\n", routine.PID(), channelName, routine.Machine().Stats().String())
		// publish message to channel
		routine.Publish(channelName, "hey there bud!")
	}, machine.GoWithTags("publish"),
		machine.GoWithTimeout(5*time.Second),
		machine.GoWithMiddlewares(
			// run every second until context cancels
			machine.Cron(time.NewTicker(1*time.Second)),
		),
	)

	m.Wait()
}

func (*Machine) Active

func (p *Machine) Active() int

Active returns current active managed goroutine count

func (*Machine) Cache

func (m *Machine) Cache() Cache

Cache returns the machine's cache implementation. One is automatically set if not provided as an Opt on machine instance creation.

func (*Machine) Cancel

func (p *Machine) Cancel()

Cancel cancels every goroutines context within the machine instance & it's children

func (*Machine) CancelRoutine

func (m *Machine) CancelRoutine(id string)

CancelRoutine cancels the context of the active routine with the given id if it exists.

func (*Machine) Close

func (m *Machine) Close()

Close completely closes the machine instance & all of it's children

func (*Machine) Go

func (m *Machine) Go(fn Func, opts ...GoOpt)

Go calls the given function in a new goroutine. it is passed information about the goroutine at runtime via the Routine interface

func (*Machine) HasRoutine

func (m *Machine) HasRoutine(id string) bool

HasRoutine returns true if the machine has a active routine with the given id

func (*Machine) ID

func (m *Machine) ID() string

ID returns the machine instance's unique id.

func (*Machine) Parent

func (m *Machine) Parent() *Machine

Parent returns the parent Machine instance if it exists and nil if not.

func (*Machine) Stats

func (m *Machine) Stats() *Stats

Stats returns Goroutine information from the machine and all of it's children

func (*Machine) Sub

func (m *Machine) Sub(opts ...Opt) *Machine

Sub returns a nested Machine instance that is dependent on the parent machine's context. It inherits the parent's pubsub/cache implementation & middlewares if none are provided Sub machine's do not inherit their parents max routine setting

func (*Machine) Tags

func (p *Machine) Tags() []string

Tags returns the machine's tags

func (*Machine) Total

func (p *Machine) Total() int

Total returns total goroutines that have been executed by the machine

func (*Machine) Wait

func (m *Machine) Wait()

Wait blocks until total active goroutine count reaches zero for the instance and all of it's children. At least one goroutine must have finished in order for wait to un-block

type Middleware

type Middleware func(fn Func) Func

Middleware is a function that wraps/modifies the behavior of a machine.Func.

func After

func After(afterFunc func(routine Routine)) Middleware

After exectues the afterFunc after the main goroutine exits.

func Before

func Before(beforeFunc func(routine Routine)) Middleware

Before exectues the beforeFunc before the main goroutine is executed.

func Cron

func Cron(ticker *time.Ticker) Middleware

Cron is a middleware that execute the function every time the ticker ticks until the goroutine's context cancels

func Decider

func Decider(deciderFunc func(routine Routine) bool) Middleware

Decider exectues the deciderFunc before the main goroutine is executed. If it returns false, the goroutine won't be executed.

func PanicRecover

func PanicRecover() Middleware

PanicRecover wraps a goroutine with a middleware the recovers from panics.

func While

func While(deciderFunc func(routine Routine) bool) Middleware

While is a middleware that will continue to execute the Func while deciderFunc() returns true. The loop breaks the first time deciderFunc() returns false or the routine's context cancels

type Opt

type Opt func(o *option)

Opt is a single option when creating a machine instance with New

func WithCache

func WithCache(cache Cache) Opt

WithCache sets the cache implementation for the machine instance. An inmemory implementation is used if none is provided.

func WithChildren

func WithChildren(children ...*Machine) Opt

WithChildren sets the machine instances children

func WithDeadline

func WithDeadline(deadline time.Time) Opt

WithDeadline is an Opt that creates the Machine's context with the given deadline.

func WithID

func WithID(id string) Opt

WithID sets the machine instances unique id. If one isn't provided, a unique id will be assigned

func WithMaxRoutines

func WithMaxRoutines(max int) Opt

WithMaxRoutines throttles goroutines at the input number. It will panic if <= zero.

func WithMiddlewares

func WithMiddlewares(middlewares ...Middleware) Opt

WithMiddlewares wraps every goroutine function executed by the machine with the given middlewares. Middlewares can be added to individual goroutines with GoWithMiddlewares

func WithPubSub

func WithPubSub(pubsub PubSub) Opt

WithPubSub sets the pubsub implementation for the machine instance. An inmemory implementation is used if none is provided.

func WithTags

func WithTags(tags ...string) Opt

WithTags sets the machine instances tags

func WithTimeout

func WithTimeout(to time.Duration) Opt

WithTimeout is an Opt that creates the Machine's context with the given timeout value

func WithValue

func WithValue(key, val interface{}) Opt

WithValue adds the k/v to the Machine's root context. It can be retrieved with context.Value() in all sub routine contexts

type PubSub

type PubSub interface {
	// Publish publishes the object to the channel by name
	Publish(channel string, obj interface{}) error
	// Subscribe subscribes to the given channel
	Subscribe(ctx context.Context, channel string, handler func(obj interface{})) error
	Close()
}

PubSub is used to asynchronously pass messages between routines.

type Routine

type Routine interface {
	// Context returns the goroutines unique context that may be used for cancellation
	Context() context.Context
	// Cancel cancels the context returned from Context()
	Cancel()
	// PID() is the goroutines unique process id
	PID() string
	// Tags() are the tags associated with the goroutine
	Tags() []string
	// Start is when the goroutine started
	Start() time.Time
	// Duration is the duration since the goroutine started
	Duration() time.Duration
	// Publish publishes the object to the given channel
	Publish(channel string, obj interface{}) error
	// Subscribe subscribes to a channel and executes the function on every message passed to it. It exits if the goroutines context is cancelled.
	Subscribe(channel string, handler func(obj interface{})) error
	// Machine returns the underlying routine's machine instance
	Machine() *Machine
}

Routine is an interface representing a goroutine

type RoutineStats

type RoutineStats struct {
	PID      string        `json:"pid"`
	Start    time.Time     `json:"start"`
	Duration time.Duration `json:"duration"`
	Tags     []string      `json:"tags"`
}

RoutineStats holds information about a single goroutine

type Stats

type Stats struct {
	ID               string         `json:"id"`
	Tags             []string       `json:"tags"`
	TotalRoutines    int            `json:"totalRoutines"`
	ActiveRoutines   int            `json:"activeRoutines"`
	Routines         []RoutineStats `json:"routines"`
	TotalChildren    int            `json:"totalChildren"`
	HasParent        bool           `json:"hasParent"`
	TotalMiddlewares int            `json:"totalMiddlewares"`
	Timeout          time.Duration  `json:"timeout"`
	Deadline         time.Time      `json:"deadline"`
	Children         []*Stats       `json:"children"`
}

Stats holds information about goroutines

func (Stats) String

func (s Stats) String() string

String prints a pretty json string of the stats