gowork

README

GoWork

GoWork is a library for the encapsulation and delivery of Work to a distributed set of Workers.

Installation

go get github.com/ryanskidmore/gowork

NOTE: Only tested with Go 1.3+

Features

• Worker Registration/Authentication
• Work Encapsulation and Tracking
• Built-in Work Queue
• Event based handler functions (e.g. run function when work complete, etc)

Usage

Server

Work

First step is to create a new work server,

ws, err := gowork.NewServer("32 character secret")

Error is only returned by this function when the secret isn't 32 characters so as long as your secret is definitely 32 characters you can safely ignore the err returned by this function.

Next, you can add parameters to be passed to your handler functions, though this is entirely optional.

ws = ws.AddParams(map[string]interface{}{"param1":"value1"})

Then, you can add handler functions to be called when certain events happen.

err = ws.NewHandler("event_id", func(e *Event, params map[string]interface{}){// function //})

This function will only return an error when the handler already exists, so in most situations you can probably ignore the err returned.

• Event ID's listed at the end of this section
• Event contains data about the event, and is of type gowork.Event. Contains data about Work, Worker, Error and timings.
• Params passed in earlier will be available in the params map (however, you will need to cast them into the appropriate type).

Now work can be created and added.

w, err := gowork.CreateWork(WorkData interface{}, Timeout int64)
ws.Add(w)

WorkData can be anything that you want to pass to the worker. Timeout is duration in seconds in integer64 type until you want the work to Timeout. Error is returned by gowork.CreateWork() when the WorkData fails to marshal into JSON format.

At this point, the work is created and added to the work queue.

The work is able to be retrieved then by calling the function Get

w, err := ws.Get("Worker ID", "Worker Authentication String")

The Worker ID and Authentication String are provided to the worker by functions below in the Workers section.

This function will return an error when:-

• The Worker ID cannot be converted to an integer
• The Work has timed out
• The Worker has failed to authenticate

When there is no work, the function will return an empty Work object and no error.

Then the work can be marshalled to be transferred to the worker (via a means of your choice) using the function Marshal() on the Work Object

workstring := w.Marshal()

To accompany this function, there is also an unmarshal function to convert the string to a work object

w := Unmarshal(workstring)

Once the worker has completed the work and passed it back to the server, it can be submitted via the Submit() function which marks the work as complete and calls the work_complete event.

ws.Submit(w)

Should you wish to get the size of the work queue, there is a function available to return this.

size := ws.QueueSize()

Event IDs

• add_handler_error
• add_work
• get_work
• get_work_empty
• get_work_error
• work_complete
• work_timeout
• worker_registered
• worker_verify
• worker_verify_error

Workers

When first receiving contact from the worker, the worker needs to be regisered. This provides the worker with an ID as well as a randomly generated string to encrypt in order to verify the secrets are the same on the client and the server, as well as to authenticate the worker.

id, clienttest := ws.Workers.Register(ws)

The client then calls the appropriate functions on these strings and returns the encrypted string (client response) for verification to the server.

authkey, err := ws.Workers.Verify(ws, id, clientresponse)

This function provides a session authentication key for the worker to use. Returns an error when:-

• Couldn't Worker ID to Int
• Failed to decrypt the client response
• Client Key is incorrect

Client

When creating a new worker, the worker must first contact the work server and get an ID and Client Test before calling any Go Work functions. Once these two strings have been obtained, the following function can be called.

Worker, err := gowork.NewWorker(Secret, Id, ClientTest)

This function will return an error when:-

• Secret is not 32 characters
• Couldn't convert Worker ID to Int
• Failed to encrypt the client test

This function will then store the client response in Worker.Verification.ClientResponse, which must then be passed to the server for verification, which will then return a session authentication key.

This key can be set within the worker by calling:-

Worker = Worker.SetAuthenticationKey(Key)

Then the worker is ready to get work, this must be done independently of the library, using the Worker.Id and Worker.SessionAuthenticationKey strings for authentication. Once the work is retrieved, it can be unmarshalled into a work object and then passed into Process

w, workdata, err := Worker.Process(w)

This function returns the original work object as well as the work data in a map of interfaces. An error will be returned when:-

• The work data cannot be unmarshalled
• The work has timed out

Once the worker has completed the work, it can be submitted using Submit

w, err := Worker.Submit(w, result, error)

This function takes the original work object, appends the result to it (string) as well as any error (string) and returns the new work object back ready for marshalling and submission to the server. An error will be returned when the work has timed out.

Contributors

Ryan Skidmore - github.com/ryanskidmore

Documentation

Index

• type ClientTest
• type Event
• type TimeStats
• type Work
  • func CreateWork(WorkData interface{}, Timeout int64) (*Work, error)
  • func Unmarshal(w string) *Work
  • func (w Work) Marshal() string
• type WorkResult
• type WorkServer
  • func NewServer(Secret string) (*WorkServer, error)
  • func (ws WorkServer) Add(w *Work)
  • func (ws WorkServer) AddParams(params map[string]interface{}) *WorkServer
  • func (ws WorkServer) Event(event_id string, event *Event)
  • func (ws WorkServer) Get(Id string, AuthenticationKey string) (*Work, error)
  • func (ws WorkServer) NewHandler(event_id string, hf func(*Event, map[string]interface{})) error
  • func (ws WorkServer) QueueSize() int
  • func (ws WorkServer) Submit(w *Work)
• type Worker
  • func NewWorker(Secret string, ID string, PlaintextVerification string) (*Worker, error)
  • func (wrk Worker) Process(w *Work) (*Work, map[string]interface{}, error)
  • func (wrk Worker) SetAuthenticationKey(key string) *Worker
  • func (wrk Worker) Submit(w *Work, ResultJSON string, Error string) (*Work, error)
• type WorkersStruct
  • func (wrs WorkersStruct) Register(ws *WorkServer) (string, string)
  • func (wrs WorkersStruct) Verify(ws *WorkServer, Id string, Response string) (string, error)

Types

type ClientTest

type ClientTest struct {
	PlaintextVerification string `json:"Verification"`
	ClientResponse        string `json:"Response"`
}

type Event

type Event struct {
	Work   *Work
	Worker *Worker
	Error  string
	Time   int64
}

type TimeStats

type TimeStats struct {
	Added    int64
	Recieved int64
	Complete int64
	Timeout  int64
}

type Work

type Work struct {
	Id       bson.ObjectId `json:"-" bson:"_id"`
	IdHex    string
	WorkJSON string
	Result   *WorkResult
	Time     *TimeStats
}

func CreateWork

func CreateWork(WorkData interface{}, Timeout int64) (*Work, error)

func Unmarshal

func Unmarshal(w string) *Work

func (Work) Marshal

func (w Work) Marshal() string

type WorkResult

type WorkResult struct {
	ResultJSON string
	Status     string
	Error      string
}

type WorkServer

type WorkServer struct {
	Queue         *lane.Queue
	Handlers      map[string]interface{}
	HandlerParams map[string]interface{}
	Workers       *WorkersStruct
}

func NewServer

func NewServer(Secret string) (*WorkServer, error)

func (WorkServer) Add

func (ws WorkServer) Add(w *Work)

func (WorkServer) AddParams

func (ws WorkServer) AddParams(params map[string]interface{}) *WorkServer

func (WorkServer) Event

func (ws WorkServer) Event(event_id string, event *Event)

func (WorkServer) Get

func (ws WorkServer) Get(Id string, AuthenticationKey string) (*Work, error)

func (WorkServer) NewHandler

func (ws WorkServer) NewHandler(event_id string, hf func(*Event, map[string]interface{})) error

func (WorkServer) QueueSize

func (ws WorkServer) QueueSize() int

func (WorkServer) Submit

func (ws WorkServer) Submit(w *Work)

type Worker

type Worker struct {
	Id                       int
	Registered               bool
	PresharedSecret          string
	SessionAuthenticationKey string
	Verification             *ClientTest
}

func NewWorker

func NewWorker(Secret string, ID string, PlaintextVerification string) (*Worker, error)

func (Worker) Process

func (wrk Worker) Process(w *Work) (*Work, map[string]interface{}, error)

func (Worker) SetAuthenticationKey

func (wrk Worker) SetAuthenticationKey(key string) *Worker

func (Worker) Submit

func (wrk Worker) Submit(w *Work, ResultJSON string, Error string) (*Work, error)

type WorkersStruct

type WorkersStruct struct {
	Members         map[int]*Worker
	PresharedSecret string
	WorkerCount     int
}

func (WorkersStruct) Register

func (wrs WorkersStruct) Register(ws *WorkServer) (string, string)

func (WorkersStruct) Verify

func (wrs WorkersStruct) Verify(ws *WorkServer, Id string, Response string) (string, error)