concurrent

Documentation

Overview

Package concurrent provides common functionality for dealing with concurrency that extends or enhances the core golang packages.

Index

• func Await(runnable Runnable, signals <-chan os.Signal) error
• func Execute(runnable Runnable) (waitGroup *sync.WaitGroup, shutdown chan struct{}, err error)
• func WaitTimeout(waitGroup *sync.WaitGroup, timeout time.Duration) bool
• type KeyValue
  • func NewKeyValue() *KeyValue
  • func (kv *KeyValue) Add(key, value interface{})
  • func (kv *KeyValue) Apply(transformer KeyValueTransformer) <-chan chan interface{}
  • func (kv *KeyValue) Delete(keys ...interface{})
  • func (kv *KeyValue) Do(operation KeyValueOperation)
  • func (kv *KeyValue) Get(key interface{}) <-chan interface{}
  • func (kv *KeyValue) Keys() <-chan chan interface{}
  • func (kv *KeyValue) Values() <-chan chan interface{}
• type KeyValueOperation
• type KeyValueOperationFunc
  • func (f KeyValueOperationFunc) Execute(storage KeyValueStorage)
• type KeyValueStorage
• type KeyValueTransformer
• type KeyValueTransformerFunc
  • func (f KeyValueTransformerFunc) Execute(key, value interface{}) interface{}
• type Runnable
• type RunnableFunc
  • func (r RunnableFunc) Run(waitGroup *sync.WaitGroup, shutdown <-chan struct{}) error
• type RunnableSet
  • func (set RunnableSet) Run(waitGroup *sync.WaitGroup, shutdown <-chan struct{}) error

Functions

func Await

func Await(runnable Runnable, signals <-chan os.Signal) error

Await uses Execute() to invoke a runnable, then waits for any traffic on a signal channel before shutting down gracefully.

func Execute

func Execute(runnable Runnable) (waitGroup *sync.WaitGroup, shutdown chan struct{}, err error)

Execute is a convenience function that creates the necessary synchronization objects and then invokes Run().

func WaitTimeout

func WaitTimeout(waitGroup *sync.WaitGroup, timeout time.Duration) bool

WaitTimeout performs a timed wait on a given sync.WaitGroup

Types

type KeyValue

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

KeyValue is a concurrent mapping of arbitrary types with a completely asynchronous API. Instances of this type must be created via NewKeyValue.

func NewKeyValue

func NewKeyValue() *KeyValue

NewKeyValue initializes and returns a distinct KeyValue instance.

func (*KeyValue) Add

func (kv *KeyValue) Add(key, value interface{})

Add asynchronously adds (or, replaces) a key/value pair.

func (*KeyValue) Apply

func (kv *KeyValue) Apply(transformer KeyValueTransformer) <-chan chan interface{}

Apply uses the given transformer to produce a result for each key/value pair in the storage. A channel of channels is returned: The channel has a buffer size of 1 and will receive another channel containing the results of applying the transformer.

func (*KeyValue) Delete

func (kv *KeyValue) Delete(keys ...interface{})

Delete asynchronously removes zero or more keys from the internal storage.

func (*KeyValue) Do

func (kv *KeyValue) Do(operation KeyValueOperation)

Do asynchronously executes a bulk operation against the internal storage. This method contends on the internal write lock.

func (*KeyValue) Get

func (kv *KeyValue) Get(key interface{}) <-chan interface{}

Get asynchronously obtains the value associated with the given key. The returned channel always receives exactly one (1) value. It will receive nil if the given key was not present in the storage.

func (*KeyValue) Keys

func (kv *KeyValue) Keys() <-chan chan interface{}

Keys is a special usage of Apply: It returns a channel which in turn receives a channel containing the keys in the internal storage.

func (*KeyValue) Values

func (kv *KeyValue) Values() <-chan chan interface{}

Values is a special usage of Apply: It returns a channel which in turn receives a channel containing the values in the internal storage.

type KeyValueOperation

type KeyValueOperation interface {
	Execute(KeyValueStorage)
}

KeyValueOperation represents an atomic operation that is allowed to mutate the storage of a KeyValue. Operations are always executed within a critical section bounded by a write lock.

type KeyValueOperationFunc

type KeyValueOperationFunc func(KeyValueStorage)

KeyValueOperationFunc is a function type that implements KeyValueOperation.

func (KeyValueOperationFunc) Execute

func (f KeyValueOperationFunc) Execute(storage KeyValueStorage)

type KeyValueStorage

type KeyValueStorage map[interface{}]interface{}

KeyValueStorage is the map type used by KeyValue. It is the type that is directly modifiable by operations.

type KeyValueTransformer

type KeyValueTransformer interface {
	Execute(key, value interface{}) interface{}
}

KeyValueTransformer is a binary operation that produces a result from a key/value pair. Transformers cannot mutate the storage of a KeyValue. Transformers are always executed within the context of a read lock. Multiple transformers can execute simultaneously.

type KeyValueTransformerFunc

type KeyValueTransformerFunc func(key, value interface{}) interface{}

KeyValueTransformerFunc is a function type that implements KeyValueTransformer.

func (KeyValueTransformerFunc) Execute

func (f KeyValueTransformerFunc) Execute(key, value interface{}) interface{}

type Runnable

type Runnable interface {
	// Run executes this operation, possibly returning an error if the operation
	// could not be started.  This method is responsible for spawning any necessary
	// goroutines and to ensure WaitGroup.Add() and WaitGroup.Done() are called appropriately.
	// Generally speaking, Run() should be idempotent.
	//
	// The supplied shutdown channel is used to signal any goroutines spawned by this
	// method that they should gracefully exit.  Callers can then use the waitGroup to
	// wait until things have been cleaned up properly.
	Run(waitGroup *sync.WaitGroup, shutdown <-chan struct{}) error
}

Runnable represents any operation that can spawn zero or more goroutines.

type RunnableFunc

type RunnableFunc func(*sync.WaitGroup, <-chan struct{}) error

RunnableFunc is a function type that implements Runnable

func (RunnableFunc) Run

func (r RunnableFunc) Run(waitGroup *sync.WaitGroup, shutdown <-chan struct{}) error

type RunnableSet

type RunnableSet []Runnable

RunnableSet is a slice type that allows grouping of operations. This type implements Runnable as well.

func (RunnableSet) Run

func (set RunnableSet) Run(waitGroup *sync.WaitGroup, shutdown <-chan struct{}) error