gearman-go/client/client.go

300 lines
6.6 KiB
Go
Raw Normal View History

// The client package helps developers connect to Gearmand, send
// jobs and fetch result.
package client
import (
2014-01-09 17:58:02 +08:00
"bufio"
2013-08-29 16:51:23 +08:00
"net"
"sync"
)
// One client connect to one server.
// Use Pool for multi-connections.
type Client struct {
2013-12-24 22:04:10 +08:00
sync.Mutex
2013-08-30 12:36:57 +08:00
net, addr, lastcall string
respHandler map[string]ResponseHandler
innerHandler map[string]ResponseHandler
in chan *Response
2013-08-30 12:36:57 +08:00
conn net.Conn
2014-01-09 17:58:02 +08:00
rw *bufio.ReadWriter
2013-08-29 16:51:23 +08:00
2013-08-30 11:20:51 +08:00
ErrorHandler ErrorHandler
}
// Return a client.
func New(network, addr string) (client *Client, err error) {
2013-08-29 16:51:23 +08:00
client = &Client{
net: network,
2013-08-30 11:20:51 +08:00
addr: addr,
respHandler: make(map[string]ResponseHandler, queueSize),
innerHandler: make(map[string]ResponseHandler, queueSize),
in: make(chan *Response, queueSize),
2013-08-29 16:51:23 +08:00
}
client.conn, err = net.Dial(client.net, client.addr)
2013-08-30 18:01:10 +08:00
if err != nil {
return
}
2014-01-09 16:16:34 +08:00
client.rw = bufio.NewReadWriter(bufio.NewReader(client.conn),
bufio.NewWriter(client.conn))
go client.readLoop()
go client.processLoop()
2013-08-29 16:51:23 +08:00
return
2013-01-14 17:59:48 +08:00
}
2013-08-29 16:51:23 +08:00
func (client *Client) write(req *request) (err error) {
var n int
buf := req.Encode()
for i := 0; i < len(buf); i += n {
2014-01-09 16:16:34 +08:00
n, err = client.rw.Write(buf[i:])
2013-08-29 16:51:23 +08:00
if err != nil {
return
}
}
2014-01-09 16:16:34 +08:00
return client.rw.Flush()
2013-01-14 17:59:48 +08:00
}
2013-08-29 16:51:23 +08:00
func (client *Client) read(length int) (data []byte, err error) {
n := 0
buf := getBuffer(bufferSize)
2013-08-29 16:51:23 +08:00
// read until data can be unpacked
for i := length; i > 0 || len(data) < minPacketLength; i -= n {
2014-01-09 16:16:34 +08:00
if n, err = client.rw.Read(buf); err != nil {
2013-08-29 16:51:23 +08:00
return
}
data = append(data, buf[0:n]...)
if n < bufferSize {
2013-08-29 16:51:23 +08:00
break
}
}
return
2013-01-14 17:59:48 +08:00
}
2013-08-29 16:51:23 +08:00
func (client *Client) readLoop() {
defer close(client.in)
var data, leftdata []byte
2013-08-29 16:51:23 +08:00
var err error
var resp *Response
2014-01-09 16:16:34 +08:00
ReadLoop:
2014-03-03 11:40:42 +08:00
for client.conn != nil {
if data, err = client.read(bufferSize); err != nil {
2014-03-03 14:45:35 +08:00
if opErr, ok := err.(*net.OpError); ok {
if opErr.Timeout() {
client.err(err)
}
if opErr.Temporary() {
continue
}
break
}
2014-03-03 14:45:35 +08:00
client.err(err)
// If it is unexpected error and the connection wasn't
// closed by Gearmand, the client should close the conection
// and reconnect to job server.
client.Close()
client.conn, err = net.Dial(client.net, client.addr)
if err != nil {
client.err(err)
2013-08-29 16:51:23 +08:00
break
}
2014-01-09 16:16:34 +08:00
client.rw = bufio.NewReadWriter(bufio.NewReader(client.conn),
bufio.NewWriter(client.conn))
2013-08-29 16:51:23 +08:00
continue
}
if len(leftdata) > 0 { // some data left for processing
2013-08-29 18:08:05 +08:00
data = append(leftdata, data...)
leftdata = nil
2013-08-29 16:51:23 +08:00
}
2014-01-09 16:16:34 +08:00
for {
l := len(data)
if l < minPacketLength { // not enough data
leftdata = data
continue ReadLoop
}
if resp, l, err = decodeResponse(data); err != nil {
leftdata = data[l:]
continue ReadLoop
} else {
client.in <- resp
}
data = data[l:]
if len(data) > 0 {
continue
}
break
2013-08-29 16:51:23 +08:00
}
}
}
func (client *Client) processLoop() {
for resp := range client.in {
switch resp.DataType {
case dtError:
if client.lastcall != "" {
resp = client.handleInner(client.lastcall, resp)
client.lastcall = ""
} else {
client.err(getError(resp.Data))
}
case dtStatusRes:
resp = client.handleInner("s"+resp.Handle, resp)
case dtJobCreated:
resp = client.handleInner("c", resp)
case dtEchoRes:
resp = client.handleInner("e", resp)
2014-01-09 16:16:34 +08:00
case dtWorkData, dtWorkWarning, dtWorkStatus:
resp = client.handleResponse(resp.Handle, resp)
case dtWorkComplete, dtWorkFail, dtWorkException:
client.handleResponse(resp.Handle, resp)
delete(client.respHandler, resp.Handle)
}
}
}
2013-08-29 16:51:23 +08:00
func (client *Client) err(e error) {
if client.ErrorHandler != nil {
client.ErrorHandler(e)
}
}
2013-12-24 22:04:10 +08:00
func (client *Client) handleResponse(key string, resp *Response) *Response {
2013-08-29 16:51:23 +08:00
if h, ok := client.respHandler[key]; ok {
h(resp)
2013-12-24 22:04:10 +08:00
return nil
2013-08-29 18:08:05 +08:00
}
2013-12-24 22:04:10 +08:00
return resp
2013-08-29 18:08:05 +08:00
}
2013-12-25 17:01:42 +08:00
func (client *Client) handleInner(key string, resp *Response) *Response {
2013-08-29 18:08:05 +08:00
if h, ok := client.innerHandler[key]; ok {
h(resp)
delete(client.innerHandler, key)
2013-12-24 22:04:10 +08:00
return nil
2013-08-29 16:51:23 +08:00
}
2013-12-24 22:04:10 +08:00
return resp
2013-01-14 17:59:48 +08:00
}
2013-01-15 17:55:44 +08:00
func (client *Client) do(funcname string, data []byte,
2013-08-30 11:20:51 +08:00
flag uint32) (handle string, err error) {
2014-03-03 11:40:42 +08:00
if client.conn == nil {
return "", ErrLostConn
}
2013-12-24 22:04:10 +08:00
var mutex sync.Mutex
mutex.Lock()
2013-08-30 11:20:51 +08:00
client.lastcall = "c"
2013-12-24 22:04:10 +08:00
client.innerHandler["c"] = func(resp *Response) {
2014-03-11 10:03:00 +08:00
defer mutex.Unlock()
if resp.DataType == dtError {
err = getError(resp.Data)
2013-08-30 11:20:51 +08:00
return
}
2013-08-29 16:51:23 +08:00
handle = resp.Handle
2013-12-24 22:04:10 +08:00
}
2013-09-22 22:58:22 +08:00
id := IdGen.Id()
req := getJob(id, []byte(funcname), data)
req.DataType = flag
2014-08-21 00:27:32 +08:00
if err = client.write(req); err != nil {
delete(client.innerHandler, "c")
client.lastcall = ""
return
}
2013-12-24 22:04:10 +08:00
mutex.Lock()
2013-08-29 16:51:23 +08:00
return
2013-01-14 17:59:48 +08:00
}
// Call the function and get a response.
// flag can be set to: JobLow, JobNormal and JobHigh
2013-01-15 17:55:44 +08:00
func (client *Client) Do(funcname string, data []byte,
2013-08-30 11:20:51 +08:00
flag byte, h ResponseHandler) (handle string, err error) {
2013-08-29 16:51:23 +08:00
var datatype uint32
switch flag {
case JobLow:
datatype = dtSubmitJobLow
case JobHigh:
datatype = dtSubmitJobHigh
2013-08-29 16:51:23 +08:00
default:
datatype = dtSubmitJob
2013-08-29 16:51:23 +08:00
}
2013-08-30 11:20:51 +08:00
handle, err = client.do(funcname, data, datatype)
2014-03-03 11:40:42 +08:00
if err == nil && h != nil {
2013-08-29 18:08:05 +08:00
client.respHandler[handle] = h
2013-08-29 16:51:23 +08:00
}
return
2013-01-14 17:59:48 +08:00
}
// Call the function in background, no response needed.
// flag can be set to: JobLow, JobNormal and JobHigh
2013-01-15 17:55:44 +08:00
func (client *Client) DoBg(funcname string, data []byte,
2013-08-30 11:20:51 +08:00
flag byte) (handle string, err error) {
2014-03-03 11:40:42 +08:00
if client.conn == nil {
return "", ErrLostConn
}
2013-08-29 16:51:23 +08:00
var datatype uint32
switch flag {
case JobLow:
datatype = dtSubmitJobLowBg
case JobHigh:
datatype = dtSubmitJobHighBg
2013-08-29 16:51:23 +08:00
default:
datatype = dtSubmitJobBg
2013-08-29 16:51:23 +08:00
}
2013-08-30 11:20:51 +08:00
handle, err = client.do(funcname, data, datatype)
2013-08-29 16:51:23 +08:00
return
}
// Get job status from job server.
2013-08-30 11:20:51 +08:00
func (client *Client) Status(handle string) (status *Status, err error) {
2014-03-03 11:40:42 +08:00
if client.conn == nil {
return nil, ErrLostConn
}
2013-12-24 22:04:10 +08:00
var mutex sync.Mutex
mutex.Lock()
2013-08-30 11:20:51 +08:00
client.lastcall = "s" + handle
2013-12-24 22:04:10 +08:00
client.innerHandler["s"+handle] = func(resp *Response) {
2014-03-11 10:03:00 +08:00
defer mutex.Unlock()
2013-08-29 18:08:05 +08:00
var err error
2014-01-09 16:16:34 +08:00
status, err = resp._status()
2013-08-29 18:08:05 +08:00
if err != nil {
client.err(err)
}
2013-12-24 22:04:10 +08:00
}
2013-09-22 22:58:22 +08:00
req := getRequest()
req.DataType = dtGetStatus
2013-09-22 22:58:22 +08:00
req.Data = []byte(handle)
client.write(req)
2013-12-24 22:04:10 +08:00
mutex.Lock()
2013-08-29 16:51:23 +08:00
return
}
// Echo.
2013-08-29 18:08:05 +08:00
func (client *Client) Echo(data []byte) (echo []byte, err error) {
2014-03-03 11:40:42 +08:00
if client.conn == nil {
return nil, ErrLostConn
}
2013-12-24 22:04:10 +08:00
var mutex sync.Mutex
mutex.Lock()
client.innerHandler["e"] = func(resp *Response) {
2013-08-29 18:08:05 +08:00
echo = resp.Data
2013-12-24 22:04:10 +08:00
mutex.Unlock()
}
2013-09-22 22:58:22 +08:00
req := getRequest()
req.DataType = dtEchoReq
2013-09-22 22:58:22 +08:00
req.Data = data
client.lastcall = "e"
2013-12-24 22:04:10 +08:00
client.write(req)
mutex.Lock()
2013-08-29 16:51:23 +08:00
return
}
// Close connection
func (client *Client) Close() (err error) {
client.Lock()
defer client.Unlock()
if client.conn != nil {
err = client.conn.Close()
client.conn = nil
}
return
}