gorkers

Examples

• Quickstart
• Multiple Go Workers
• Passing Fields
• Timeout Worker
• Deadline Worker

Getting Started

Import

import (
    "github.com/guilhem/gorkers"
)

Create a worker function 👷

Starting from 0.2, gorkers use go1.18 generics ❤️

Define, your in and out type and create your worker:

work := func(ctx context.Context, in string, out chan<- int) error {
    // work iteration here
}

Create runner 🚶

runner := gorkers.NewRunner(ctx, work, numberOfWorkers, sizeOfBuffer)

• numberOfWorkers is the number of parallel workers that can be running at the same time
• sizeOfBuffer is the buffer size of input. If stopped, a runner can lose it's buffer.

Start runner 🏃

if err := runner.Start(); err != nil {
    // error management
}

.Start() can return an error if beforeFunc is in error.

Send work to worker

runner.Send("Hello World")

Send accepts an interface. So send it anything you want.

Wait for the worker to finish

runner.Wait()

.Wait() lock any new .Send() and block until all jobs are finished.

runner.Close()

Use .Close() to prevent any new job to be spawn and sending a context cancellation to any worker.

Stop on errors

StopWhenError is a special function to stop Runner when 1 task return an error.

runner.AfterFunc(gorkers.StopWhenError)
runner.Start()

Log errors

By default errors are ignored. To manage them you can use AfterFunc this way:

logf := func(ctx context.Context, in interface{}, err error) error {
    if err != nil {
        log.Printf("err: %s", err)
    }
    return nil
}
runner.AfterFunc(logf)

Working With Multiple Workers

Passing work form one worker to the next

By using the InFrom method you can tell workerTwo to accept output from workerOne

runnerOne := gorkers.NewRunner(ctx, work1, 100, 100)
runnerTwo := gorkers.NewRunner(ctx, work2, 100, 100).InFrom(workerOne)

runnerOne.Start()
runnerTwo.Start()

runnerOne.Wait().Stop()
runnerTwo.Wait().Stop()

Accepting output from multiple workers

It is possible to accept output from more than one worker but it is up to you to determine what is coming from which worker. (They will send on the same channel.)

runnerOne := gorkers.gewRunner(ctx, NewMyWorker(), 100, 100)
runnerTwo := gorkers.NewRunner(ctx, NewMyWorkerTwo(), 100, 100)
runnerThree := gorkers.NewRunner(ctx, NewMyWorkerThree(), 100, 100).InFrom(workerOne, workerTwo)

Options

Timeout

If your workers needs to stop at a deadline or you just need to have a timeout use the SetTimeout or SetDeadline methods. (These must be in place before setting the workers off to work.)

 // Setting a timeout of 2 seconds
 runner.SetWorkerTimeout(2 * time.Second)

.SetWorkerTimeout() is a timeout for a worker instance to finish.

Deadline

// Setting a deadline of 4 hours from now
runner.SetDeadline(time.Now().Add(4 \* time.Hour))

.SetDeadline() is a limit for runner to finish.