yuxh/gearman-go
1
0
Fork 1
Code Issues Pull Requests Releases Wiki Activity

0.2 refactoring begining

Browse Source
This commit is contained in:
Xing Xing 2013-08-29 16:51:23 +08:00
parent 2002bb1804
commit e5c30068cd
13 changed files with 798 additions and 712 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

350
client/client.go
View File

@ -9,24 +9,9 @@ import (
"io" "io"
"net" "net"
"sync" "sync"
"time" "github.com/mikespook/golib/idgen"
"bytes"
"strconv"
"github.com/mikespook/gearman-go/common"
) )
var (
IdGen IdGenerator
)
func init() {
IdGen = NewObjectId()
}
// Status handler
// handle, known, running, numerator, denominator
type StatusHandler func(string, bool, bool, uint64, uint64)
/* /*
The client side api for gearman The client side api for gearman
@ -36,47 +21,35 @@ handle := c.Do("foobar", []byte("data here"), JOB_LOW | JOB_BG)
*/ */
type Client struct { type Client struct {
ErrHandler common.ErrorHandler net, addr string
TimeOut time.Duration respHandler map[string]ResponseHandler
createdHandler ResponseHandler
in chan []byte in chan []byte
out chan *Job
created chan string
echo chan []byte
status chan *Status
jobhandlers map[string]JobHandler
isConn bool isConn bool
conn net.Conn conn net.Conn
addr string
mutex sync.RWMutex mutex sync.RWMutex
ErrorHandler ErrorHandler
IdGen idgen.IdGen
} }
// Create a new client. // Create a new client.
// Connect to "addr" through "network" // Connect to "addr" through "network"
// Eg. // Eg.
// client, err := client.New("127.0.0.1:4730") // client, err := client.New("127.0.0.1:4730")
func New(addr string) (client *Client, err error) { func New(net, addr string) (client *Client, err error) {
client = &Client{ client = &Client{
created: make(chan string, common.QUEUE_SIZE), net: net,
echo: make(chan []byte, common.QUEUE_SIZE),
status: make(chan *Status, common.QUEUE_SIZE),
jobhandlers: make(map[string]JobHandler, common.QUEUE_SIZE),
in: make(chan []byte, common.QUEUE_SIZE),
out: make(chan *Job, common.QUEUE_SIZE),
addr: addr, addr: addr,
TimeOut: time.Second, respHandler: make(map[string]ResponseHandler, QUEUE_SIZE),
in: make(chan []byte, QUEUE_SIZE),
} }
if err = client.connect(); err != nil { if err = client.connect(); err != nil {
return return
} }
client.isConn = true client.isConn = true
go client.inLoop() go client.readLoop()
go client.outLoop() go client.processLoop()
return return
} }
@ -84,13 +57,14 @@ func New(addr string) (client *Client, err error) {
// //
func (client *Client) connect() (err error) { func (client *Client) connect() (err error) {
client.conn, err = net.Dial(common.NETWORK, client.addr) client.conn, err = net.Dial(client.net, client.addr)
return return
} }
// Internal write // Internal write
func (client *Client) write(buf []byte) (err error) { func (client *Client) write(req *request) (err error) {
var n int var n int
buf := req.Encode()
for i := 0; i < len(buf); i += n { for i := 0; i < len(buf); i += n {
n, err = client.conn.Write(buf[i:]) n, err = client.conn.Write(buf[i:])
if err != nil { if err != nil {
@ -101,201 +75,109 @@ func (client *Client) write(buf []byte) (err error) {
} }
// read length bytes from the socket // read length bytes from the socket
func (client *Client) readData(length int) (data []byte, err error) { func (client *Client) read(length int) (data []byte, err error) {
n := 0 n := 0
buf := make([]byte, common.BUFFER_SIZE) buf := getBuffer(BUFFER_SIZE)
// read until data can be unpacked // read until data can be unpacked
for i := length; i > 0 || len(data) < common.PACKET_LEN; i -= n { for i := length; i > 0 || len(data) < MIN_PACKET_LEN; i -= n {
if n, err = client.conn.Read(buf); err != nil { if n, err = client.conn.Read(buf); err != nil {
if !client.isConn { if !client.isConn {
return nil, common.ErrConnClosed err = ErrConnClosed
return
} }
if err == io.EOF && n == 0 { if err == io.EOF && n == 0 {
if data == nil { if data == nil {
err = common.ErrConnection err = ErrConnection
return
} }
return data, nil
} }
return return
} }
data = append(data, buf[0:n]...) data = append(data, buf[0:n]...)
if n < common.BUFFER_SIZE { if n < BUFFER_SIZE {
break break
} }
} }
return return
} }
// unpack data // read data from socket
func (client *Client) unpack(data []byte) ([]byte, int, bool) { func (client *Client) readLoop() {
tl := len(data)
start := 0
for i := 0; i < tl+1-common.PACKET_LEN; i++ {
if start+common.PACKET_LEN > tl { // too few data to unpack, read more
return nil, common.PACKET_LEN, false
}
if string(data[start:start+4]) == common.RES_STR {
l := int(common.BytesToUint32([4]byte{data[start+8],
data[start+9], data[start+10], data[start+11]}))
total := l + common.PACKET_LEN
if total == tl { // data is what we want
return data, common.PACKET_LEN, true
} else if total < tl { // data[:total] is what we want, data[total:] is the more
client.in <- data[total:]
data = data[start:total]
return data, common.PACKET_LEN, true
} else { // ops! It won't be possible.
return nil, total - tl, false
}
} else { // flag was not found, move to next step
start++
}
}
return nil, common.PACKET_LEN, false
}
// Internal read
func (client *Client) read() (rel []byte, err error) {
var data []byte var data []byte
ok := false var err error
l := common.PACKET_LEN for client.isConn {
for !ok { if data, err = client.read(BUFFER_SIZE); err != nil {
inlen := len(client.in) if err == ErrConnClosed {
if inlen > 0 {
// in queue is not empty
for i := 0; i < inlen; i++ {
data = append(data, <-client.in...)
}
} else {
var d []byte
d, err = client.readData(l)
if err != nil {
return
}
data = append(data, d...)
}
rel, l, ok = client.unpack(data)
}
return
}
// out loop
func (client *Client) outLoop() {
for job := range client.out {
if err := client.write(job.Encode()); err != nil {
client.err(err)
}
}
}
// in loop
func (client *Client) inLoop() {
defer common.DisablePanic()
for {
rel, err := client.read()
if err != nil {
if err == common.ErrConnection {
client.Close()
}
if err != common.ErrConnClosed {
client.err(err)
}
break break
} }
job, err := decodeJob(rel)
if err != nil {
client.err(err) client.err(err)
continue continue
//break
} }
switch job.DataType { client.in <- data
case common.ERROR: }
_, err := common.GetError(job.Data) close(client.in)
}
// decode data & process it
func (client *Client) processLoop() {
var resp *response
var l int
var err error
var data, leftdata []byte
for data = range client.in {
l = len(data)
if len(leftdata) > 0 { // some data left for processing
data = append(leftdata, data ...)
}
if l < MIN_PACKET_LEN { // not enough data
leftdata = data
continue
}
if resp, l, err = decodeResponse(data); err != nil {
client.err(err) client.err(err)
case common.WORK_DATA, common.WORK_WARNING, common.WORK_STATUS, continue
common.WORK_COMPLETE, common.WORK_FAIL, common.WORK_EXCEPTION: }
client.handleJob(job) switch resp.DataType {
case common.ECHO_RES: case WORK_DATA, WORK_WARNING, WORK_STATUS, WORK_COMPLETE,
client.handleEcho(job) WORK_FAIL, WORK_EXCEPTION:
case common.JOB_CREATED: client.handleResponse(string(resp.Handle), resp)
client.handleCreated(job) }
case common.STATUS_RES: if len(data) > l {
client.handleStatus(job) leftdata = data[l:]
default:
break
} }
} }
} }
// error handler // error handler
func (client *Client) err (e error) { func (client *Client) err(e error) {
if client.ErrHandler != nil { if client.ErrorHandler != nil {
client.ErrHandler(e) client.ErrorHandler(e)
} }
} }
// job handler // job handler
func (client *Client) handleJob(job *Job) { func (client *Client) handleResponse(key string, resp *response) {
client.mutex.RLock() client.mutex.RLock()
defer client.mutex.RUnlock() defer client.mutex.RUnlock()
if h, ok := client.jobhandlers[job.Handle]; ok { if h, ok := client.respHandler[key]; ok {
h(job) h(resp)
delete(client.jobhandlers, job.Handle) delete(client.respHandler, string(resp.Handle))
} }
} }
func (client *Client) handleEcho(job *Job) {
client.echo <- job.Data
}
func (client *Client) handleCreated(job *Job) {
client.created <- string(job.Data)
}
// status handler
func (client *Client) handleStatus(job *Job) {
data := bytes.SplitN(job.Data, []byte{'\x00'}, 5)
if len(data) != 5 {
client.err(common.Errorf("Invalid data: %V", job.Data))
return
}
status := &Status{}
status.Handle = string(data[0])
status.Known = (data[1][0] == '1')
status.Running = (data[2][0] == '1')
var err error
status.Numerator, err = strconv.ParseUint(string(data[3]), 10, 0)
if err != nil {
client.err(common.Errorf("Invalid Integer: %s", data[3]))
return
}
status.Denominator, err = strconv.ParseUint(string(data[4]), 10, 0)
if err != nil {
client.err(common.Errorf("Invalid Integer: %s", data[4]))
return
}
client.status <- status
}
// Send the job to job server.
func (client *Client) writeJob(job *Job) {
client.out <- job
}
// Internal do // Internal do
func (client *Client) do(funcname string, data []byte, func (client *Client) do(funcname string, data []byte,
flag uint32, id string) (handle string) { flag uint32) (handle []byte) {
l := len(funcname) + len(id) + len(data) + 2 req := getJob(funcname, client.IdGen.Id().(string), data)
rel := make([]byte, 0, l) client.mutex.Lock()
rel = append(rel, []byte(funcname)...) // len(funcname) defer client.mutex.Unlock()
rel = append(rel, '\x00') // 1 Byte client.write(req)
rel = append(rel, []byte(id)...) // len(uid) var wg sync.WaitGroup
rel = append(rel, '\x00') // 1 Byte wg.Add(1)
rel = append(rel, data...) // len(data) client.createdHandler = func(resp *response) {
client.writeJob(newJob(common.REQ, flag, rel)) defer wg.Done()
// Waiting for JOB_CREATED handle = resp.Handle
handle = <-client.created }
wg.Wait()
return return
} }
@ -304,65 +186,67 @@ flag uint32, id string) (handle string) {
// Do the function. // Do the function.
// funcname is a string with function name. // funcname is a string with function name.
// data is encoding to byte array. // data is encoding to byte array.
// flag set the job type, include running level: JOB_LOW, JOB_NORMAL, JOB_HIGH, // flag set the job type, include running level: JOB_LOW, JOB_NORMAL, JOB_HIGH
// and if it is background job: JOB_BG.
// JOB_LOW | JOB_BG means the job is running with low level in background.
func (client *Client) Do(funcname string, data []byte, func (client *Client) Do(funcname string, data []byte,
flag byte, jobhandler JobHandler) (handle string) { flag byte, h ResponseHandler) (handle []byte) {
var datatype uint32 var datatype uint32
switch flag { switch flag {
case JOB_LOW : case JOB_LOW:
datatype = common.SUBMIT_JOB_LOW datatype = SUBMIT_JOB_LOW
case JOB_HIGH : case JOB_HIGH:
datatype = common.SUBMIT_JOB_HIGH datatype = SUBMIT_JOB_HIGH
default: default:
datatype = common.SUBMIT_JOB datatype = SUBMIT_JOB
} }
id := IdGen.Id() handle = client.do(funcname, data, datatype)
client.mutex.Lock() client.mutex.Lock()
defer client.mutex.Unlock() defer client.mutex.Unlock()
handle = client.do(funcname, data, datatype, id) if h != nil {
if jobhandler != nil { client.respHandler[string(handle)] = h
client.jobhandlers[handle] = jobhandler
} }
return return
} }
// Do the function at background.
// funcname is a string with function name.
// data is encoding to byte array.
// flag set the job type, include running level: JOB_LOW, JOB_NORMAL, JOB_HIGH
func (client *Client) DoBg(funcname string, data []byte, func (client *Client) DoBg(funcname string, data []byte,
flag byte) (handle string) { flag byte) (handle []byte) {
var datatype uint32 var datatype uint32
switch flag { switch flag {
case JOB_LOW : case JOB_LOW:
datatype = common.SUBMIT_JOB_LOW_BG datatype = SUBMIT_JOB_LOW_BG
case JOB_HIGH : case JOB_HIGH:
datatype = common.SUBMIT_JOB_HIGH_BG datatype = SUBMIT_JOB_HIGH_BG
default: default:
datatype = common.SUBMIT_JOB_BG datatype = SUBMIT_JOB_BG
} }
handle = client.do(funcname, data, datatype, "") handle = client.do(funcname, data, datatype)
return return
} }
// Get job status from job server. // Get job status from job server.
// !!!Not fully tested.!!! // !!!Not fully tested.!!!
func (client *Client) Status(handle string, timeout time.Duration) (status *Status, err error) { func (client *Client) Status(handle []byte, h ResponseHandler) (err error) {
client.writeJob(newJob(common.REQ, common.GET_STATUS, []byte(handle))) req := getRequest()
select { req.DataType = GET_STATUS
case status = <-client.status: req.Data = handle
case <-time.After(timeout): client.write(req)
err = common.ErrTimeOut if h != nil {
client.respHandler["status-" + string(handle)] = h
} }
return return
} }
// Send a something out, get the samething back. // Send a something out, get the samething back.
func (client *Client) Echo(data []byte, timeout time.Duration) (r []byte, err error) { func (client *Client) Echo(data []byte, h ResponseHandler) (err error) {
client.writeJob(newJob(common.REQ, common.ECHO_REQ, data)) req := getRequest()
select { req.DataType = ECHO_REQ
case r = <-client.echo: req.Data = data
case <-time.After(timeout): client.write(req)
err = common.ErrTimeOut if h != nil {
client.respHandler["echo"] = h
} }
return return
} }
@ -370,11 +254,5 @@ func (client *Client) Echo(data []byte, timeout time.Duration) (r []byte, err er
// Close // Close
func (client *Client) Close() (err error) { func (client *Client) Close() (err error) {
client.isConn = false client.isConn = false
close(client.in) return client.conn.Close()
close(client.out)
close(client.echo)
close(client.created)
close(client.status)
return client.conn.Close();
} }

5
client/client_test.go
View File

@ -1,8 +1,8 @@
package client package client
import ( import (
"time"
"testing" "testing"
"time"
) )
var client *Client var client *Client
@ -72,7 +72,7 @@ func TestClientStatus(t *testing.T) {
return return
} }
handle := client.Do("Delay5sec", []byte("abcdef"), JOB_LOW, nil); handle := client.Do("Delay5sec", []byte("abcdef"), JOB_LOW, nil)
s2, err := client.Status(handle, time.Second) s2, err := client.Status(handle, time.Second)
if err != nil { if err != nil {
t.Error(err) t.Error(err)
@ -88,7 +88,6 @@ func TestClientStatus(t *testing.T) {
} }
} }
func TestClientClose(t *testing.T) { func TestClientClose(t *testing.T) {
if err := client.Close(); err != nil { if err := client.Close(); err != nil {
t.Error(err) t.Error(err)

75
client/common.go Normal file
View File

@ -0,0 +1,75 @@
// Copyright 2011 - 2012 Xing Xing <mikespook@gmail.com>.
// All rights reserved.
// Use of this source code is governed by a MIT
// license that can be found in the LICENSE file.
package client
const (
NETWORK = "tcp"
// queue size
QUEUE_SIZE = 8
// read buffer size
BUFFER_SIZE = 1024
// min packet length
MIN_PACKET_LEN = 12
// \x00REQ
REQ = 5391697
REQ_STR = "\x00REQ"
// \x00RES
RES = 5391699
RES_STR = "\x00RES"
// package data type
CAN_DO = 1
CANT_DO = 2
RESET_ABILITIES = 3
PRE_SLEEP = 4
NOOP = 6
JOB_CREATED = 8
GRAB_JOB = 9
NO_JOB = 10
JOB_ASSIGN = 11
WORK_STATUS = 12
WORK_COMPLETE = 13
WORK_FAIL = 14
GET_STATUS = 15
ECHO_REQ = 16
ECHO_RES = 17
ERROR = 19
STATUS_RES = 20
SET_CLIENT_ID = 22
CAN_DO_TIMEOUT = 23
WORK_EXCEPTION = 25
WORK_DATA = 28
WORK_WARNING = 29
GRAB_JOB_UNIQ = 30
JOB_ASSIGN_UNIQ = 31
SUBMIT_JOB = 7
SUBMIT_JOB_BG = 18
SUBMIT_JOB_HIGH = 21
SUBMIT_JOB_HIGH_BG = 32
SUBMIT_JOB_LOW = 33
SUBMIT_JOB_LOW_BG = 34
)
const (
// Job type
// JOB_NORMAL | JOB_BG means a normal level job run in background
// normal level
JOB_NORMAL = 0
// background job
JOB_BG = 1
// low level
JOB_LOW = 2
// high level
JOB_HIGH = 4
)
func getBuffer(l int) (buf []byte) {
// TODO add byte buffer pool
buf = make([]byte, l)
return
}

48
client/common_test.go Normal file
View File

@ -0,0 +1,48 @@
package client
import (
"bytes"
"testing"
)
var (
testCase = map[uint32][4]byte{
0: [...]byte{0, 0, 0, 0},
1: [...]byte{0, 0, 0, 1},
256: [...]byte{0, 0, 1, 0},
256 * 256: [...]byte{0, 1, 0, 0},
256 * 256 * 256: [...]byte{1, 0, 0, 0},
256*256*256 + 256*256 + 256 + 1: [...]byte{1, 1, 1, 1},
4294967295: [...]byte{0xFF, 0xFF, 0xFF, 0xFF},
}
)
func TestUint32ToBytes(t *testing.T) {
for k, v := range testCase {
b := Uint32ToBytes(k)
if bytes.Compare(b[:], v[:]) != 0 {
t.Errorf("%v was expected, but %v was got", v, b)
}
}
}
func TestBytesToUint32s(t *testing.T) {
for k, v := range testCase {
u := BytesToUint32([4]byte(v))
if u != k {
t.Errorf("%v was expected, but %v was got", k, u)
}
}
}
func BenchmarkByteToUnit32(b *testing.B) {
for i := 0; i < b.N; i++ {
BytesToUint32([4]byte{0xF, 0xF, 0xF, 0xF})
}
}
func BenchmarkUint32ToByte(b *testing.B) {
for i := 0; i < b.N; i++ {
Uint32ToBytes(123456)
}
}

49
client/error.go Normal file
View File

@ -0,0 +1,49 @@
// Copyright 2011 - 2012 Xing Xing <mikespook@gmail.com>.
// All rights reserved.
// Use of this source code is governed by a MIT
// license that can be found in the LICENSE file.
package client
import (
"bytes"
"errors"
"fmt"
"syscall"
"strconv"
)
var (
ErrJobTimeOut = errors.New("Do a job time out")
ErrInvalidData = errors.New("Invalid data")
ErrWorkWarning = errors.New("Work warning")
ErrWorkFail = errors.New("Work fail")
ErrWorkException = errors.New("Work exeption")
ErrDataType = errors.New("Invalid data type")
ErrOutOfCap = errors.New("Out of the capability")
ErrNotConn = errors.New("Did not connect to job server")
ErrFuncNotFound = errors.New("The function was not found")
ErrConnection = errors.New("Connection error")
ErrNoActiveAgent = errors.New("No active agent")
ErrTimeOut = errors.New("Executing time out")
ErrUnknown = errors.New("Unknown error")
ErrConnClosed = errors.New("Connection closed")
)
func DisablePanic() { recover() }
// Extract the error message
func GetError(data []byte) (eno syscall.Errno, err error) {
rel := bytes.SplitN(data, []byte{'\x00'}, 2)
if len(rel) != 2 {
err = fmt.Errorf("Not a error data: %V", data)
return
}
var n uint64
if n, err = strconv.ParseUint(string(rel[0]), 10, 0); err != nil {
return
}
eno = syscall.Errno(n)
err = errors.New(string(rel[1]))
return
}

10
client/handler.go
View File

@ -1,4 +1,10 @@
package client package client
// Job handler // Response handler
type JobHandler func(*Job) type ResponseHandler func(*response)
// Error handler
type ErrorHandler func(error)
// Status handler
// handle, known, running, numerator, denominator
type StatusHandler func(string, bool, bool, uint64, uint64)

6
client/id.go
View File

@ -1,9 +1,9 @@
package client package client
import ( import (
"strconv"
"labix.org/v2/mgo/bson"
"github.com/mikespook/golib/autoinc" "github.com/mikespook/golib/autoinc"
"labix.org/v2/mgo/bson"
"strconv"
) )
type IdGenerator interface { type IdGenerator interface {
@ -11,7 +11,7 @@ type IdGenerator interface {
} }
// ObjectId // ObjectId
type objectId struct {} type objectId struct{}
func (id *objectId) Id() string { func (id *objectId) Id() string {
return bson.NewObjectId().Hex() return bson.NewObjectId().Hex()

142
client/job.go
View File

@ -1,142 +0,0 @@
// Copyright 2011 Xing Xing <mikespook@gmail.com>.
// All rights reserved.
// Use of this source code is governed by a MIT
// license that can be found in the LICENSE file.
package client
import (
"bytes"
"github.com/mikespook/gearman-go/common"
)
const (
// Job type
// JOB_NORMAL | JOB_BG means a normal level job run in background
// normal level
JOB_NORMAL = 0
// background job
JOB_BG = 1
// low level
JOB_LOW = 2
// high level
JOB_HIGH = 4
)
// An error handler
type ErrorHandler func(error)
// Client side job
type Job struct {
Data []byte
Handle, UniqueId string
magicCode, DataType uint32
}
// Create a new job
func newJob(magiccode, datatype uint32, data []byte) (job *Job) {
return &Job{magicCode: magiccode,
DataType: datatype,
Data: data}
}
// Decode a job from byte slice
func decodeJob(data []byte) (job *Job, err error) {
if len(data) < 12 {
return nil, common.Errorf("Invalid data: %V", data)
}
datatype := common.BytesToUint32([4]byte{data[4], data[5], data[6], data[7]})
l := common.BytesToUint32([4]byte{data[8], data[9], data[10], data[11]})
if len(data[12:]) != int(l) {
return nil, common.Errorf("Invalid data: %V", data)
}
data = data[12:]
var handle string
switch datatype {
case common.WORK_DATA, common.WORK_WARNING, common.WORK_STATUS,
common.WORK_COMPLETE, common.WORK_FAIL, common.WORK_EXCEPTION:
i := bytes.IndexByte(data, '\x00')
if i != -1 {
handle = string(data[:i])
data = data[i + 1:]
}
}
return &Job{magicCode: common.RES,
DataType: datatype,
Data: data,
Handle: handle}, nil
}
// Encode a job to byte slice
func (job *Job) Encode() (data []byte) {
l := len(job.Data)
tl := l + 12
data = make([]byte, tl)
magiccode := common.Uint32ToBytes(job.magicCode)
datatype := common.Uint32ToBytes(job.DataType)
datalength := common.Uint32ToBytes(uint32(l))
for i := 0; i < tl; i ++ {
switch {
case i < 4:
data[i] = magiccode[i]
case i < 8:
data[i] = datatype[i - 4]
case i < 12:
data[i] = datalength[i - 8]
default:
data[i] = job.Data[i - 12]
}
}
// Alternative
/*
data = append(data, magiccode[:] ...)
data = append(data, datatype[:] ...)
data = append(data, datalength[:] ...)
data = append(data, job.Data ...)
*/
return
}
// Extract the job's result.
func (job *Job) Result() (data []byte, err error) {
switch job.DataType {
case common.WORK_FAIL:
job.Handle = string(job.Data)
return nil, common.ErrWorkFail
case common.WORK_EXCEPTION:
err = common.ErrWorkException
fallthrough
case common.WORK_COMPLETE:
s := bytes.SplitN(job.Data, []byte{'\x00'}, 2)
if len(s) != 2 {
return nil, common.Errorf("Invalid data: %V", job.Data)
}
job.Handle = string(s[0])
data = s[1]
default:
err = common.ErrDataType
}
return
}
// Extract the job's update
func (job *Job) Update() (data []byte, err error) {
if job.DataType != common.WORK_DATA && job.DataType != common.WORK_WARNING {
err = common.ErrDataType
return
}
s := bytes.SplitN(job.Data, []byte{'\x00'}, 2)
if len(s) != 2 {
err = common.ErrInvalidData
return
}
if job.DataType == common.WORK_WARNING {
err = common.ErrWorkWarning
}
job.Handle = string(s[0])
data = s[1]
return
}

31
client/pool.go
View File

@ -6,11 +6,9 @@
package client package client
import ( import (
"sync"
"time"
"errors" "errors"
"math/rand" "math/rand"
"github.com/mikespook/gearman-go/common" "sync"
) )
const ( const (
@ -29,7 +27,7 @@ type poolClient struct {
type SelectionHandler func(map[string]*poolClient, string) string type SelectionHandler func(map[string]*poolClient, string) string
func SelectWithRate(pool map[string]*poolClient, func SelectWithRate(pool map[string]*poolClient,
last string) (addr string) { last string) (addr string) {
total := 0 total := 0
for _, item := range pool { for _, item := range pool {
total += item.Rate total += item.Rate
@ -41,23 +39,21 @@ last string) (addr string) {
} }
func SelectRandom(pool map[string]*poolClient, func SelectRandom(pool map[string]*poolClient,
last string) (addr string) { last string) (addr string) {
r := rand.Intn(len(pool)) r := rand.Intn(len(pool))
i := 0 i := 0
for k, _ := range pool { for k, _ := range pool {
if r == i { if r == i {
return k return k
} }
i ++ i++
} }
return last return last
} }
type Pool struct { type Pool struct {
SelectionHandler SelectionHandler SelectionHandler SelectionHandler
ErrHandler common.ErrorHandler ErrorHandler ErrorHandler
clients map[string]*poolClient clients map[string]*poolClient
last string last string
@ -74,7 +70,7 @@ func NewPool() (pool *Pool) {
} }
// Add a server with rate. // Add a server with rate.
func (pool *Pool) Add(addr string, rate int) (err error) { func (pool *Pool) Add(net, addr string, rate int) (err error) {
pool.mutex.Lock() pool.mutex.Lock()
defer pool.mutex.Unlock() defer pool.mutex.Unlock()
var item *poolClient var item *poolClient
@ -83,7 +79,7 @@ func (pool *Pool) Add(addr string, rate int) (err error) {
item.Rate = rate item.Rate = rate
} else { } else {
var client *Client var client *Client
client, err = New(addr) client, err = New(net, addr)
item = &poolClient{Client: client, Rate: rate} item = &poolClient{Client: client, Rate: rate}
pool.clients[addr] = item pool.clients[addr] = item
} }
@ -98,7 +94,7 @@ func (pool *Pool) Remove(addr string) {
} }
func (pool *Pool) Do(funcname string, data []byte, func (pool *Pool) Do(funcname string, data []byte,
flag byte, h JobHandler) (addr, handle string) { flag byte, h ResponseHandler) (addr string, handle []byte) {
client := pool.selectServer() client := pool.selectServer()
handle = client.Do(funcname, data, flag, h) handle = client.Do(funcname, data, flag, h)
addr = client.addr addr = client.addr
@ -106,7 +102,7 @@ flag byte, h JobHandler) (addr, handle string) {
} }
func (pool *Pool) DoBg(funcname string, data []byte, func (pool *Pool) DoBg(funcname string, data []byte,
flag byte) (addr, handle string) { flag byte) (addr string, handle []byte) {
client := pool.selectServer() client := pool.selectServer()
handle = client.DoBg(funcname, data, flag) handle = client.DoBg(funcname, data, flag)
addr = client.addr addr = client.addr
@ -115,9 +111,9 @@ flag byte) (addr, handle string) {
// Get job status from job server. // Get job status from job server.
// !!!Not fully tested.!!! // !!!Not fully tested.!!!
func (pool *Pool) Status(addr, handle string, timeout time.Duration) (status *Status, err error) { func (pool *Pool) Status(addr string, handle []byte, h ResponseHandler) (err error) {
if client, ok := pool.clients[addr]; ok { if client, ok := pool.clients[addr]; ok {
status, err = client.Status(handle, timeout) err = client.Status(handle, h)
} else { } else {
err = ErrNotFound err = ErrNotFound
} }
@ -125,7 +121,7 @@ func (pool *Pool) Status(addr, handle string, timeout time.Duration) (status *St
} }
// Send a something out, get the samething back. // Send a something out, get the samething back.
func (pool *Pool) Echo(addr string, data []byte, timeout time.Duration) (r []byte, err error) { func (pool *Pool) Echo(addr string, data []byte, h ResponseHandler) (r []byte, err error) {
var client *poolClient var client *poolClient
if addr == "" { if addr == "" {
client = pool.selectServer() client = pool.selectServer()
@ -136,7 +132,7 @@ func (pool *Pool) Echo(addr string, data []byte, timeout time.Duration) (r []byt
return return
} }
} }
r, err = client.Echo(data, timeout) err = client.Echo(data, h)
return return
} }
@ -149,6 +145,7 @@ func (pool *Pool) Close() (err map[string]error) {
return return
} }
// selecting server
func (pool *Pool) selectServer() (client *poolClient) { func (pool *Pool) selectServer() (client *poolClient) {
for client == nil { for client == nil {
addr := pool.SelectionHandler(pool.clients, pool.last) addr := pool.SelectionHandler(pool.clients, pool.last)

4
client/pool_test.go
View File

@ -1,8 +1,8 @@
package client package client
import ( import (
"time"
"testing" "testing"
"time"
) )
var ( var (
@ -79,7 +79,7 @@ func TestPoolStatus(t *testing.T) {
t.Errorf("The job (%s) shouldn't be running.", s1.Handle) t.Errorf("The job (%s) shouldn't be running.", s1.Handle)
} }
addr, handle := pool.Do("Delay5sec", []byte("abcdef"), JOB_LOW, nil); addr, handle := pool.Do("Delay5sec", []byte("abcdef"), JOB_LOW, nil)
s2, err := pool.Status(addr, handle, time.Second) s2, err := pool.Status(addr, handle, time.Second)
if err != nil { if err != nil {
t.Error(err) t.Error(err)

47
client/request.go Normal file
View File

@ -0,0 +1,47 @@
// Copyright 2013 Xing Xing <mikespook@gmail.com>.
// All rights reserved.
// Use of this source code is governed by a MIT
// license that can be found in the LICENSE file.
package client
import (
"encoding/binary"
)
// request
type request struct {
DataType uint32
Data []byte
}
// Encode a Request to byte slice
func (req *request) Encode() (data []byte) {
l := len(req.Data) // length of data
tl := l + 12 // add 12 bytes head
data = getBuffer(tl)
copy(data[:4], REQ_STR)
binary.BigEndian.PutUint32(data[4:8], req.DataType)
binary.BigEndian.PutUint32(data[8:12], uint32(l))
copy(data[12:], req.Data)
return
}
func getRequest() (req *request) {
// TODO add a pool
req = &request{}
return
}
func getJob(funcname, id string, data []byte) (req *request) {
req = getRequest()
a := len(funcname)
b := len(id)
c := len(data)
l := a + b + c + 2
req.Data = getBuffer(l)
copy(req.Data[0:a], funcname)
copy(req.Data[a+1:a+b+1], []byte(id))
copy(req.Data[a+b+1:a+b+c+1], data)
return
}

129
client/response.go Normal file
View File

@ -0,0 +1,129 @@
// Copyright 2011 Xing Xing <mikespook@gmail.com>.
// All rights reserved.
// Use of this source code is governed by a MIT
// license that can be found in the LICENSE file.
package client
import (
"fmt"
"bytes"
"strconv"
"encoding/binary"
)
// response
type response struct {
DataType uint32
Data, Handle []byte
UID string
}
// Extract the Response's result.
// if data == nil, err != nil, then worker failing to execute job
// if data != nil, err != nil, then worker has a exception
// if data != nil, err == nil, then worker complate job
// after calling this method, the Response.Handle will be filled
func (resp *response) Result() (data []byte, err error) {
switch resp.DataType {
case WORK_FAIL:
resp.Handle = resp.Data
err = ErrWorkFail
return
case WORK_EXCEPTION:
err = ErrWorkException
fallthrough
case WORK_COMPLETE:
s := bytes.SplitN(resp.Data, []byte{'\x00'}, 2)
if len(s) != 2 {
err = fmt.Errorf("Invalid data: %V", resp.Data)
return
}
resp.Handle = s[0]
data = s[1]
default:
err = ErrDataType
}
return
}
// Extract the job's update
func (resp *response) Update() (data []byte, err error) {
if resp.DataType != WORK_DATA &&
resp.DataType != WORK_WARNING {
err = ErrDataType
return
}
s := bytes.SplitN(resp.Data, []byte{'\x00'}, 2)
if len(s) != 2 {
err = ErrInvalidData
return
}
if resp.DataType == WORK_WARNING {
err = ErrWorkWarning
}
resp.Handle = s[0]
data = s[1]
return
}
// Decode a job from byte slice
func decodeResponse(data []byte) (resp *response, l int, err error) {
if len(data) < MIN_PACKET_LEN { // valid package should not less 12 bytes
err = fmt.Errorf("Invalid data: %V", data)
return
}
dl := int(binary.BigEndian.Uint32(data[8:12]))
dt := data[MIN_PACKET_LEN:dl+MIN_PACKET_LEN]
if len(dt) != int(dl) { // length not equal
err = fmt.Errorf("Invalid data: %V", data)
return
}
resp = getResponse()
resp.DataType = binary.BigEndian.Uint32(data[4:8])
switch resp.DataType {
case WORK_DATA, WORK_WARNING, WORK_STATUS,
WORK_COMPLETE, WORK_FAIL, WORK_EXCEPTION:
s := bytes.SplitN(data, []byte{'\x00'}, 2)
if len(s) >= 2 {
resp.Handle = s[0]
resp.Data = s[1]
} else {
err = fmt.Errorf("Invalid data: %V", data)
return
}
}
l = len(resp.Data) + MIN_PACKET_LEN
return
}
// status handler
func (resp *response) Status() (status *Status, err error) {
data := bytes.SplitN(resp.Data, []byte{'\x00'}, 5)
if len(data) != 5 {
err = fmt.Errorf("Invalid data: %V", resp.Data)
return
}
status = &Status{}
status.Handle = data[0]
status.Known = (data[1][0] == '1')
status.Running = (data[2][0] == '1')
status.Numerator, err = strconv.ParseUint(string(data[3]), 10, 0)
if err != nil {
err = fmt.Errorf("Invalid Integer: %s", data[3])
return
}
status.Denominator, err = strconv.ParseUint(string(data[4]), 10, 0)
if err != nil {
err = fmt.Errorf("Invalid Integer: %s", data[4])
return
}
return
}
func getResponse() (resp *response) {
// TODO add a pool
resp = &response{}
return
}

2
client/status.go
View File

@ -1,7 +1,7 @@
package client package client
type Status struct { type Status struct {
Handle string Handle []byte
Known, Running bool Known, Running bool
Numerator, Denominator uint64 Numerator, Denominator uint64
} }