Compare commits
No commits in common. "master" and "0.2-dev" have entirely different histories.
@ -1,6 +1,6 @@
|
|||||||
language: go
|
language: go
|
||||||
go:
|
go:
|
||||||
- 1.4
|
- 1.2
|
||||||
|
|
||||||
before_install:
|
before_install:
|
||||||
- sudo apt-get remove -y gearman-job-server
|
- sudo apt-get remove -y gearman-job-server
|
||||||
|
82
README.md
82
README.md
@ -37,46 +37,41 @@ Usage
|
|||||||
|
|
||||||
## Worker
|
## Worker
|
||||||
|
|
||||||
```go
|
// Limit number of concurrent jobs execution. Use worker.Unlimited (0) if you want no limitation.
|
||||||
// Limit number of concurrent jobs execution.
|
w := worker.New(worker.OneByOne)
|
||||||
// Use worker.Unlimited (0) if you want no limitation.
|
w.ErrHandler = func(e error) {
|
||||||
w := worker.New(worker.OneByOne)
|
log.Println(e)
|
||||||
w.ErrHandler = func(e error) {
|
}
|
||||||
log.Println(e)
|
w.AddServer("127.0.0.1:4730")
|
||||||
}
|
// Use worker.Unlimited (0) if you want no timeout
|
||||||
w.AddServer("127.0.0.1:4730")
|
w.AddFunc("ToUpper", ToUpper, worker.Unlimited)
|
||||||
// Use worker.Unlimited (0) if you want no timeout
|
// This will give a timeout of 5 seconds
|
||||||
w.AddFunc("ToUpper", ToUpper, worker.Unlimited)
|
w.AddFunc("ToUpperTimeOut5", ToUpper, 5)
|
||||||
// This will give a timeout of 5 seconds
|
if err := w.Ready(); err != nil {
|
||||||
w.AddFunc("ToUpperTimeOut5", ToUpper, 5)
|
log.Fatal(err)
|
||||||
|
return
|
||||||
if err := w.Ready(); err != nil {
|
}
|
||||||
log.Fatal(err)
|
go w.Work()
|
||||||
return
|
|
||||||
}
|
|
||||||
go w.Work()
|
|
||||||
```
|
|
||||||
|
|
||||||
## Client
|
## Client
|
||||||
|
|
||||||
```go
|
// ...
|
||||||
// ...
|
c, err := client.New("tcp4", "127.0.0.1:4730")
|
||||||
c, err := client.New("tcp4", "127.0.0.1:4730")
|
// ... error handling
|
||||||
// ... error handling
|
defer c.Close()
|
||||||
defer c.Close()
|
c.ErrorHandler = func(e error) {
|
||||||
c.ErrorHandler = func(e error) {
|
log.Println(e)
|
||||||
log.Println(e)
|
}
|
||||||
}
|
echo := []byte("Hello\x00 world")
|
||||||
echo := []byte("Hello\x00 world")
|
echomsg, err := c.Echo(echo)
|
||||||
echomsg, err := c.Echo(echo)
|
// ... error handling
|
||||||
// ... error handling
|
log.Println(string(echomsg))
|
||||||
log.Println(string(echomsg))
|
jobHandler := func(job *client.Job) {
|
||||||
jobHandler := func(resp *client.Response) {
|
log.Printf("%s", job.Data)
|
||||||
log.Printf("%s", resp.Data)
|
}
|
||||||
}
|
handle, err := c.Do("ToUpper", echo, client.JOB_NORMAL, jobHandler)
|
||||||
handle, err := c.Do("ToUpper", echo, client.JobNormal, jobHandler)
|
// ...
|
||||||
// ...
|
|
||||||
```
|
|
||||||
|
|
||||||
Branches
|
Branches
|
||||||
========
|
========
|
||||||
@ -92,15 +87,10 @@ __Use at your own risk!__
|
|||||||
Contributors
|
Contributors
|
||||||
============
|
============
|
||||||
|
|
||||||
Great thanks to all of you for your support and interest!
|
|
||||||
|
|
||||||
(_Alphabetic order_)
|
(_Alphabetic order_)
|
||||||
|
|
||||||
* [Alex Zylman](https://github.com/azylman)
|
* [Alex Zylman](https://github.com/azylman)
|
||||||
* [C.R. Kirkwood-Watts](https://github.com/kirkwood)
|
|
||||||
* [Damian Gryski](https://github.com/dgryski)
|
* [Damian Gryski](https://github.com/dgryski)
|
||||||
* [Gabriel Cristian Alecu](https://github.com/AzuraMeta)
|
|
||||||
* [Graham Barr](https://github.com/gbarr)
|
|
||||||
* [Ingo Oeser](https://github.com/nightlyone)
|
* [Ingo Oeser](https://github.com/nightlyone)
|
||||||
* [jake](https://github.com/jbaikge)
|
* [jake](https://github.com/jbaikge)
|
||||||
* [Joe Higton](https://github.com/draxil)
|
* [Joe Higton](https://github.com/draxil)
|
||||||
@ -108,13 +98,9 @@ Great thanks to all of you for your support and interest!
|
|||||||
* [Kevin Darlington](https://github.com/kdar)
|
* [Kevin Darlington](https://github.com/kdar)
|
||||||
* [miraclesu](https://github.com/miraclesu)
|
* [miraclesu](https://github.com/miraclesu)
|
||||||
* [Paul Mach](https://github.com/paulmach)
|
* [Paul Mach](https://github.com/paulmach)
|
||||||
* [Randall McPherson](https://github.com/rlmcpherson)
|
|
||||||
* [Sam Grimee](https://github.com/sgrimee)
|
* [Sam Grimee](https://github.com/sgrimee)
|
||||||
|
* suchj
|
||||||
Maintainer
|
* [Xing Xing](http://mikespook.com) <mikespook@gmail.com> [@Twitter](http://twitter.com/mikespook)
|
||||||
==========
|
|
||||||
|
|
||||||
* [Xing Xing](http://mikespook.com) <<mikespook@gmail.com>> [@Twitter](http://twitter.com/mikespook)
|
|
||||||
|
|
||||||
Open Source - MIT Software License
|
Open Source - MIT Software License
|
||||||
==================================
|
==================================
|
||||||
|
230
client/client.go
230
client/client.go
@ -6,11 +6,6 @@ import (
|
|||||||
"bufio"
|
"bufio"
|
||||||
"net"
|
"net"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
|
||||||
)
|
|
||||||
|
|
||||||
var (
|
|
||||||
DefaultTimeout time.Duration = time.Second
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// One client connect to one server.
|
// One client connect to one server.
|
||||||
@ -18,63 +13,24 @@ var (
|
|||||||
type Client struct {
|
type Client struct {
|
||||||
sync.Mutex
|
sync.Mutex
|
||||||
|
|
||||||
net, addr string
|
net, addr, lastcall string
|
||||||
innerHandler *responseHandlerMap
|
respHandler map[string]ResponseHandler
|
||||||
in chan *Response
|
innerHandler map[string]ResponseHandler
|
||||||
conn net.Conn
|
in chan *Response
|
||||||
rw *bufio.ReadWriter
|
conn net.Conn
|
||||||
|
rw *bufio.ReadWriter
|
||||||
ResponseTimeout time.Duration // response timeout for do()
|
|
||||||
|
|
||||||
ErrorHandler ErrorHandler
|
ErrorHandler ErrorHandler
|
||||||
}
|
}
|
||||||
|
|
||||||
type responseHandlerMap struct {
|
// Return a client.
|
||||||
sync.Mutex
|
|
||||||
holder map[string]handledResponse
|
|
||||||
}
|
|
||||||
|
|
||||||
type handledResponse struct {
|
|
||||||
internal ResponseHandler // internal handler, always non-nil
|
|
||||||
external ResponseHandler // handler passed in from (*Client).Do, sometimes nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func newResponseHandlerMap() *responseHandlerMap {
|
|
||||||
return &responseHandlerMap{holder: make(map[string]handledResponse, queueSize)}
|
|
||||||
}
|
|
||||||
|
|
||||||
func (r *responseHandlerMap) remove(key string) {
|
|
||||||
r.Lock()
|
|
||||||
delete(r.holder, key)
|
|
||||||
r.Unlock()
|
|
||||||
}
|
|
||||||
|
|
||||||
func (r *responseHandlerMap) getAndRemove(key string) (handledResponse, bool) {
|
|
||||||
r.Lock()
|
|
||||||
rh, b := r.holder[key]
|
|
||||||
delete(r.holder, key)
|
|
||||||
r.Unlock()
|
|
||||||
return rh, b
|
|
||||||
}
|
|
||||||
|
|
||||||
func (r *responseHandlerMap) putWithExternalHandler(key string, internal, external ResponseHandler) {
|
|
||||||
r.Lock()
|
|
||||||
r.holder[key] = handledResponse{internal: internal, external: external}
|
|
||||||
r.Unlock()
|
|
||||||
}
|
|
||||||
|
|
||||||
func (r *responseHandlerMap) put(key string, rh ResponseHandler) {
|
|
||||||
r.putWithExternalHandler(key, rh, nil)
|
|
||||||
}
|
|
||||||
|
|
||||||
// New returns a client.
|
|
||||||
func New(network, addr string) (client *Client, err error) {
|
func New(network, addr string) (client *Client, err error) {
|
||||||
client = &Client{
|
client = &Client{
|
||||||
net: network,
|
net: network,
|
||||||
addr: addr,
|
addr: addr,
|
||||||
innerHandler: newResponseHandlerMap(),
|
respHandler: make(map[string]ResponseHandler, queueSize),
|
||||||
in: make(chan *Response, queueSize),
|
innerHandler: make(map[string]ResponseHandler, queueSize),
|
||||||
ResponseTimeout: DefaultTimeout,
|
in: make(chan *Response, queueSize),
|
||||||
}
|
}
|
||||||
client.conn, err = net.Dial(client.net, client.addr)
|
client.conn, err = net.Dial(client.net, client.addr)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@ -148,7 +104,6 @@ ReadLoop:
|
|||||||
}
|
}
|
||||||
if len(leftdata) > 0 { // some data left for processing
|
if len(leftdata) > 0 { // some data left for processing
|
||||||
data = append(leftdata, data...)
|
data = append(leftdata, data...)
|
||||||
leftdata = nil
|
|
||||||
}
|
}
|
||||||
for {
|
for {
|
||||||
l := len(data)
|
l := len(data)
|
||||||
@ -172,25 +127,27 @@ ReadLoop:
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (client *Client) processLoop() {
|
func (client *Client) processLoop() {
|
||||||
rhandlers := map[string]ResponseHandler{}
|
|
||||||
for resp := range client.in {
|
for resp := range client.in {
|
||||||
switch resp.DataType {
|
switch resp.DataType {
|
||||||
case dtError:
|
case dtError:
|
||||||
client.err(getError(resp.Data))
|
if client.lastcall != "" {
|
||||||
case dtStatusRes:
|
resp = client.handleInner(client.lastcall, resp)
|
||||||
client.handleInner("s"+resp.Handle, resp, nil)
|
client.lastcall = ""
|
||||||
case dtJobCreated:
|
} else {
|
||||||
client.handleInner("c", resp, rhandlers)
|
client.err(getError(resp.Data))
|
||||||
case dtEchoRes:
|
|
||||||
client.handleInner("e", resp, nil)
|
|
||||||
case dtWorkData, dtWorkWarning, dtWorkStatus:
|
|
||||||
if cb := rhandlers[resp.Handle]; cb != nil {
|
|
||||||
cb(resp)
|
|
||||||
}
|
}
|
||||||
|
case dtStatusRes:
|
||||||
|
resp = client.handleInner("s"+resp.Handle, resp)
|
||||||
|
case dtJobCreated:
|
||||||
|
resp = client.handleInner("c", resp)
|
||||||
|
case dtEchoRes:
|
||||||
|
resp = client.handleInner("e", resp)
|
||||||
|
case dtWorkData, dtWorkWarning, dtWorkStatus:
|
||||||
|
resp = client.handleResponse(resp.Handle, resp)
|
||||||
case dtWorkComplete, dtWorkFail, dtWorkException:
|
case dtWorkComplete, dtWorkFail, dtWorkException:
|
||||||
if cb := rhandlers[resp.Handle]; cb != nil {
|
resp = client.handleResponse(resp.Handle, resp)
|
||||||
cb(resp)
|
if resp != nil {
|
||||||
delete(rhandlers, resp.Handle)
|
delete(client.respHandler, resp.Handle)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -202,54 +159,44 @@ func (client *Client) err(e error) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (client *Client) handleInner(key string, resp *Response, rhandlers map[string]ResponseHandler) {
|
func (client *Client) handleResponse(key string, resp *Response) *Response {
|
||||||
if h, ok := client.innerHandler.getAndRemove(key); ok {
|
if h, ok := client.respHandler[key]; ok {
|
||||||
if h.external != nil && resp.Handle != "" {
|
h(resp)
|
||||||
rhandlers[resp.Handle] = h.external
|
return nil
|
||||||
}
|
|
||||||
h.internal(resp)
|
|
||||||
}
|
}
|
||||||
|
return resp
|
||||||
}
|
}
|
||||||
|
|
||||||
type handleOrError struct {
|
func (client *Client) handleInner(key string, resp *Response) *Response {
|
||||||
handle string
|
if h, ok := client.innerHandler[key]; ok {
|
||||||
err error
|
h(resp)
|
||||||
|
delete(client.innerHandler, key)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return resp
|
||||||
}
|
}
|
||||||
|
|
||||||
func (client *Client) do(funcname string, data []byte,
|
func (client *Client) do(funcname string, data []byte,
|
||||||
flag uint32, h ResponseHandler, id string) (handle string, err error) {
|
flag uint32) (handle string, err error) {
|
||||||
if len(id) == 0 {
|
|
||||||
return "", ErrInvalidId
|
|
||||||
}
|
|
||||||
if client.conn == nil {
|
if client.conn == nil {
|
||||||
return "", ErrLostConn
|
return "", ErrLostConn
|
||||||
}
|
}
|
||||||
var result = make(chan handleOrError, 1)
|
var mutex sync.Mutex
|
||||||
client.Lock()
|
mutex.Lock()
|
||||||
defer client.Unlock()
|
client.lastcall = "c"
|
||||||
client.innerHandler.putWithExternalHandler("c", func(resp *Response) {
|
client.innerHandler["c"] = func(resp *Response) {
|
||||||
|
defer mutex.Unlock()
|
||||||
if resp.DataType == dtError {
|
if resp.DataType == dtError {
|
||||||
err = getError(resp.Data)
|
err = getError(resp.Data)
|
||||||
result <- handleOrError{"", err}
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
handle = resp.Handle
|
handle = resp.Handle
|
||||||
result <- handleOrError{handle, nil}
|
}
|
||||||
}, h)
|
id := IdGen.Id()
|
||||||
req := getJob(id, []byte(funcname), data)
|
req := getJob(id, []byte(funcname), data)
|
||||||
req.DataType = flag
|
req.DataType = flag
|
||||||
if err = client.write(req); err != nil {
|
client.write(req)
|
||||||
client.innerHandler.remove("c")
|
mutex.Lock()
|
||||||
return
|
|
||||||
}
|
|
||||||
var timer = time.After(client.ResponseTimeout)
|
|
||||||
select {
|
|
||||||
case ret := <-result:
|
|
||||||
return ret.handle, ret.err
|
|
||||||
case <-timer:
|
|
||||||
client.innerHandler.remove("c")
|
|
||||||
return "", ErrLostConn
|
|
||||||
}
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -257,7 +204,19 @@ func (client *Client) do(funcname string, data []byte,
|
|||||||
// flag can be set to: JobLow, JobNormal and JobHigh
|
// flag can be set to: JobLow, JobNormal and JobHigh
|
||||||
func (client *Client) Do(funcname string, data []byte,
|
func (client *Client) Do(funcname string, data []byte,
|
||||||
flag byte, h ResponseHandler) (handle string, err error) {
|
flag byte, h ResponseHandler) (handle string, err error) {
|
||||||
handle, err = client.DoWithId(funcname, data, flag, h, IdGen.Id())
|
var datatype uint32
|
||||||
|
switch flag {
|
||||||
|
case JobLow:
|
||||||
|
datatype = dtSubmitJobLow
|
||||||
|
case JobHigh:
|
||||||
|
datatype = dtSubmitJobHigh
|
||||||
|
default:
|
||||||
|
datatype = dtSubmitJob
|
||||||
|
}
|
||||||
|
handle, err = client.do(funcname, data, datatype)
|
||||||
|
if err == nil && h != nil {
|
||||||
|
client.respHandler[handle] = h
|
||||||
|
}
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -265,25 +224,38 @@ func (client *Client) Do(funcname string, data []byte,
|
|||||||
// flag can be set to: JobLow, JobNormal and JobHigh
|
// flag can be set to: JobLow, JobNormal and JobHigh
|
||||||
func (client *Client) DoBg(funcname string, data []byte,
|
func (client *Client) DoBg(funcname string, data []byte,
|
||||||
flag byte) (handle string, err error) {
|
flag byte) (handle string, err error) {
|
||||||
handle, err = client.DoBgWithId(funcname, data, flag, IdGen.Id())
|
if client.conn == nil {
|
||||||
|
return "", ErrLostConn
|
||||||
|
}
|
||||||
|
var datatype uint32
|
||||||
|
switch flag {
|
||||||
|
case JobLow:
|
||||||
|
datatype = dtSubmitJobLowBg
|
||||||
|
case JobHigh:
|
||||||
|
datatype = dtSubmitJobHighBg
|
||||||
|
default:
|
||||||
|
datatype = dtSubmitJobBg
|
||||||
|
}
|
||||||
|
handle, err = client.do(funcname, data, datatype)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// Status gets job status from job server.
|
// Get job status from job server.
|
||||||
func (client *Client) Status(handle string) (status *Status, err error) {
|
func (client *Client) Status(handle string) (status *Status, err error) {
|
||||||
if client.conn == nil {
|
if client.conn == nil {
|
||||||
return nil, ErrLostConn
|
return nil, ErrLostConn
|
||||||
}
|
}
|
||||||
var mutex sync.Mutex
|
var mutex sync.Mutex
|
||||||
mutex.Lock()
|
mutex.Lock()
|
||||||
client.innerHandler.put("s"+handle, func(resp *Response) {
|
client.lastcall = "s" + handle
|
||||||
|
client.innerHandler["s"+handle] = func(resp *Response) {
|
||||||
defer mutex.Unlock()
|
defer mutex.Unlock()
|
||||||
var err error
|
var err error
|
||||||
status, err = resp._status()
|
status, err = resp._status()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
client.err(err)
|
client.err(err)
|
||||||
}
|
}
|
||||||
})
|
}
|
||||||
req := getRequest()
|
req := getRequest()
|
||||||
req.DataType = dtGetStatus
|
req.DataType = dtGetStatus
|
||||||
req.Data = []byte(handle)
|
req.Data = []byte(handle)
|
||||||
@ -299,13 +271,14 @@ func (client *Client) Echo(data []byte) (echo []byte, err error) {
|
|||||||
}
|
}
|
||||||
var mutex sync.Mutex
|
var mutex sync.Mutex
|
||||||
mutex.Lock()
|
mutex.Lock()
|
||||||
client.innerHandler.put("e", func(resp *Response) {
|
client.innerHandler["e"] = func(resp *Response) {
|
||||||
echo = resp.Data
|
echo = resp.Data
|
||||||
mutex.Unlock()
|
mutex.Unlock()
|
||||||
})
|
}
|
||||||
req := getRequest()
|
req := getRequest()
|
||||||
req.DataType = dtEchoReq
|
req.DataType = dtEchoReq
|
||||||
req.Data = data
|
req.Data = data
|
||||||
|
client.lastcall = "e"
|
||||||
client.write(req)
|
client.write(req)
|
||||||
mutex.Lock()
|
mutex.Lock()
|
||||||
return
|
return
|
||||||
@ -321,40 +294,3 @@ func (client *Client) Close() (err error) {
|
|||||||
}
|
}
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// Call the function and get a response.
|
|
||||||
// flag can be set to: JobLow, JobNormal and JobHigh
|
|
||||||
func (client *Client) DoWithId(funcname string, data []byte,
|
|
||||||
flag byte, h ResponseHandler, id string) (handle string, err error) {
|
|
||||||
var datatype uint32
|
|
||||||
switch flag {
|
|
||||||
case JobLow:
|
|
||||||
datatype = dtSubmitJobLow
|
|
||||||
case JobHigh:
|
|
||||||
datatype = dtSubmitJobHigh
|
|
||||||
default:
|
|
||||||
datatype = dtSubmitJob
|
|
||||||
}
|
|
||||||
handle, err = client.do(funcname, data, datatype, h, id)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
// Call the function in background, no response needed.
|
|
||||||
// flag can be set to: JobLow, JobNormal and JobHigh
|
|
||||||
func (client *Client) DoBgWithId(funcname string, data []byte,
|
|
||||||
flag byte, id string) (handle string, err error) {
|
|
||||||
if client.conn == nil {
|
|
||||||
return "", ErrLostConn
|
|
||||||
}
|
|
||||||
var datatype uint32
|
|
||||||
switch flag {
|
|
||||||
case JobLow:
|
|
||||||
datatype = dtSubmitJobLowBg
|
|
||||||
case JobHigh:
|
|
||||||
datatype = dtSubmitJobHighBg
|
|
||||||
default:
|
|
||||||
datatype = dtSubmitJobBg
|
|
||||||
}
|
|
||||||
handle, err = client.do(funcname, data, datatype, nil, id)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
@ -1,39 +1,16 @@
|
|||||||
package client
|
package client
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"crypto/md5"
|
|
||||||
"encoding/hex"
|
|
||||||
"errors"
|
|
||||||
"flag"
|
|
||||||
"fmt"
|
|
||||||
"os"
|
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
TestStr = "Hello world"
|
TestStr = "Hello world"
|
||||||
)
|
)
|
||||||
|
|
||||||
var (
|
var client *Client
|
||||||
client *Client
|
|
||||||
runIntegrationTests bool
|
|
||||||
)
|
|
||||||
|
|
||||||
func TestMain(m *testing.M) {
|
|
||||||
integrationsTestFlag := flag.Bool("integration", false, "Run the integration tests (in addition to the unit tests)")
|
|
||||||
flag.Parse()
|
|
||||||
if integrationsTestFlag != nil {
|
|
||||||
runIntegrationTests = *integrationsTestFlag
|
|
||||||
}
|
|
||||||
code := m.Run()
|
|
||||||
os.Exit(code)
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestClientAddServer(t *testing.T) {
|
func TestClientAddServer(t *testing.T) {
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
t.Log("Add local server 127.0.0.1:4730")
|
t.Log("Add local server 127.0.0.1:4730")
|
||||||
var err error
|
var err error
|
||||||
if client, err = New(Network, "127.0.0.1:4730"); err != nil {
|
if client, err = New(Network, "127.0.0.1:4730"); err != nil {
|
||||||
@ -45,9 +22,6 @@ func TestClientAddServer(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestClientEcho(t *testing.T) {
|
func TestClientEcho(t *testing.T) {
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
echo, err := client.Echo([]byte(TestStr))
|
echo, err := client.Echo([]byte(TestStr))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Error(err)
|
t.Error(err)
|
||||||
@ -60,9 +34,6 @@ func TestClientEcho(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestClientDoBg(t *testing.T) {
|
func TestClientDoBg(t *testing.T) {
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
handle, err := client.DoBg("ToUpper", []byte("abcdef"), JobLow)
|
handle, err := client.DoBg("ToUpper", []byte("abcdef"), JobLow)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Error(err)
|
t.Error(err)
|
||||||
@ -75,46 +46,7 @@ func TestClientDoBg(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestClientDoBgWithId(t *testing.T) {
|
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
data := []byte("abcdef")
|
|
||||||
hash := md5.Sum(data)
|
|
||||||
id := hex.EncodeToString(hash[:])
|
|
||||||
handle, err := client.DoBgWithId("ToUpper", data, JobLow, id)
|
|
||||||
if err != nil {
|
|
||||||
t.Error(err)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if handle == "" {
|
|
||||||
t.Error("Handle is empty.")
|
|
||||||
} else {
|
|
||||||
t.Log(handle)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestClientDoBgWithIdFailsIfNoId(t *testing.T) {
|
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
data := []byte("abcdef")
|
|
||||||
id := ""
|
|
||||||
_, err := client.DoBgWithId("ToUpper", data, JobLow, id)
|
|
||||||
if err == nil {
|
|
||||||
t.Error("Expecting error")
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if err.Error() != "Invalid ID" {
|
|
||||||
t.Error(fmt.Sprintf("Expecting \"Invalid ID\" error, got %s.", err.Error()))
|
|
||||||
return
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestClientDo(t *testing.T) {
|
func TestClientDo(t *testing.T) {
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
jobHandler := func(job *Response) {
|
jobHandler := func(job *Response) {
|
||||||
str := string(job.Data)
|
str := string(job.Data)
|
||||||
if str == "ABCDEF" {
|
if str == "ABCDEF" {
|
||||||
@ -137,202 +69,7 @@ func TestClientDo(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestClientDoWithId(t *testing.T) {
|
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
jobHandler := func(job *Response) {
|
|
||||||
str := string(job.Data)
|
|
||||||
if str == "ABCDEF" {
|
|
||||||
t.Log(str)
|
|
||||||
} else {
|
|
||||||
t.Errorf("Invalid data: %s", job.Data)
|
|
||||||
}
|
|
||||||
return
|
|
||||||
}
|
|
||||||
data := []byte("abcdef")
|
|
||||||
hash := md5.Sum(data)
|
|
||||||
id := hex.EncodeToString(hash[:])
|
|
||||||
handle, err := client.DoWithId("ToUpper", data,
|
|
||||||
JobLow, jobHandler, id)
|
|
||||||
if err != nil {
|
|
||||||
t.Error(err)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if handle == "" {
|
|
||||||
t.Error("Handle is empty.")
|
|
||||||
} else {
|
|
||||||
t.Log(handle)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestClientDoWithIdFailsIfNoId(t *testing.T) {
|
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
jobHandler := func(job *Response) {
|
|
||||||
str := string(job.Data)
|
|
||||||
if str == "ABCDEF" {
|
|
||||||
t.Log(str)
|
|
||||||
} else {
|
|
||||||
t.Errorf("Invalid data: %s", job.Data)
|
|
||||||
}
|
|
||||||
return
|
|
||||||
}
|
|
||||||
data := []byte("abcdef")
|
|
||||||
id := ""
|
|
||||||
_, err := client.DoWithId("ToUpper", data,
|
|
||||||
JobLow, jobHandler, id)
|
|
||||||
if err == nil {
|
|
||||||
t.Error("Expecting error")
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if err.Error() != "Invalid ID" {
|
|
||||||
t.Error(fmt.Sprintf("Expecting \"Invalid ID\" error, got %s.", err.Error()))
|
|
||||||
return
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestClientDoWithIdCheckSameHandle(t *testing.T) {
|
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
jobHandler := func(job *Response) {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
data := []byte("{productId:123,categoryId:1}")
|
|
||||||
id := "123"
|
|
||||||
handle1, err := client.DoWithId("PublishProduct", data,
|
|
||||||
JobLow, jobHandler, id)
|
|
||||||
if err != nil {
|
|
||||||
t.Error(err)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if handle1 == "" {
|
|
||||||
t.Error("Handle is empty.")
|
|
||||||
} else {
|
|
||||||
t.Log(handle1)
|
|
||||||
}
|
|
||||||
|
|
||||||
handle2, err := client.DoWithId("PublishProduct", data,
|
|
||||||
JobLow, jobHandler, id)
|
|
||||||
if err != nil {
|
|
||||||
t.Error(err)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if handle2 == "" {
|
|
||||||
t.Error("Handle is empty.")
|
|
||||||
} else {
|
|
||||||
t.Log(handle2)
|
|
||||||
}
|
|
||||||
|
|
||||||
if handle1 != handle2 {
|
|
||||||
t.Error("expecting the same handle when using the same id on the same Job name")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestClientDoWithIdCheckDifferentHandleOnDifferentJobs(t *testing.T) {
|
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
jobHandler := func(job *Response) {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
data := []byte("{productId:123}")
|
|
||||||
id := "123"
|
|
||||||
handle1, err := client.DoWithId("PublishProduct", data,
|
|
||||||
JobLow, jobHandler, id)
|
|
||||||
if err != nil {
|
|
||||||
t.Error(err)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if handle1 == "" {
|
|
||||||
t.Error("Handle is empty.")
|
|
||||||
} else {
|
|
||||||
t.Log(handle1)
|
|
||||||
}
|
|
||||||
|
|
||||||
handle2, err := client.DoWithId("DeleteProduct", data,
|
|
||||||
JobLow, jobHandler, id)
|
|
||||||
if err != nil {
|
|
||||||
t.Error(err)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if handle2 == "" {
|
|
||||||
t.Error("Handle is empty.")
|
|
||||||
} else {
|
|
||||||
t.Log(handle2)
|
|
||||||
}
|
|
||||||
|
|
||||||
if handle1 == handle2 {
|
|
||||||
t.Error("expecting different handles because there are different job names")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestClientMultiDo(t *testing.T) {
|
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
|
|
||||||
// This integration test requires that examples/pl/worker_multi.pl be running.
|
|
||||||
//
|
|
||||||
// Test invocation is:
|
|
||||||
// go test -integration -timeout 10s -run '^TestClient(AddServer|MultiDo)$'
|
|
||||||
//
|
|
||||||
// Send 1000 requests to go through all race conditions
|
|
||||||
const nreqs = 1000
|
|
||||||
errCh := make(chan error)
|
|
||||||
gotCh := make(chan string, nreqs)
|
|
||||||
|
|
||||||
olderrh := client.ErrorHandler
|
|
||||||
client.ErrorHandler = func(e error) { errCh <- e }
|
|
||||||
client.ResponseTimeout = 5 * time.Second
|
|
||||||
defer func() { client.ErrorHandler = olderrh }()
|
|
||||||
|
|
||||||
nextJobCh := make(chan struct{})
|
|
||||||
defer close(nextJobCh)
|
|
||||||
go func() {
|
|
||||||
for range nextJobCh {
|
|
||||||
start := time.Now()
|
|
||||||
handle, err := client.Do("PerlToUpper", []byte("abcdef"), JobNormal, func(r *Response) { gotCh <- string(r.Data) })
|
|
||||||
if err == ErrLostConn && time.Since(start) > client.ResponseTimeout {
|
|
||||||
errCh <- errors.New("Impossible 'lost conn', deadlock bug detected")
|
|
||||||
} else if err != nil {
|
|
||||||
errCh <- err
|
|
||||||
}
|
|
||||||
if handle == "" {
|
|
||||||
errCh <- errors.New("Handle is empty.")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
for i := 0; i < nreqs; i++ {
|
|
||||||
select {
|
|
||||||
case err := <-errCh:
|
|
||||||
t.Fatal(err)
|
|
||||||
case nextJobCh <- struct{}{}:
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
remaining := nreqs
|
|
||||||
for remaining > 0 {
|
|
||||||
select {
|
|
||||||
case err := <-errCh:
|
|
||||||
t.Fatal(err)
|
|
||||||
case got := <-gotCh:
|
|
||||||
if got != "ABCDEF" {
|
|
||||||
t.Error("Unexpected response from PerlDoUpper: ", got)
|
|
||||||
}
|
|
||||||
remaining--
|
|
||||||
t.Logf("%d response remaining", remaining)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestClientStatus(t *testing.T) {
|
func TestClientStatus(t *testing.T) {
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
status, err := client.Status("handle not exists")
|
status, err := client.Status("handle not exists")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Error(err)
|
t.Error(err)
|
||||||
@ -368,9 +105,6 @@ func TestClientStatus(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestClientClose(t *testing.T) {
|
func TestClientClose(t *testing.T) {
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
if err := client.Close(); err != nil {
|
if err := client.Close(); err != nil {
|
||||||
t.Error(err)
|
t.Error(err)
|
||||||
}
|
}
|
||||||
|
@ -5,7 +5,7 @@ const (
|
|||||||
// queue size
|
// queue size
|
||||||
queueSize = 8
|
queueSize = 8
|
||||||
// read buffer size
|
// read buffer size
|
||||||
bufferSize = 8192
|
bufferSize = 1024
|
||||||
// min packet length
|
// min packet length
|
||||||
minPacketLength = 12
|
minPacketLength = 12
|
||||||
|
|
||||||
@ -51,7 +51,6 @@ const (
|
|||||||
dtSubmitJobLowBg = 34
|
dtSubmitJobLowBg = 34
|
||||||
|
|
||||||
WorkComplate = dtWorkComplete
|
WorkComplate = dtWorkComplete
|
||||||
WorkComplete = dtWorkComplete
|
|
||||||
WorkData = dtWorkData
|
WorkData = dtWorkData
|
||||||
WorkStatus = dtWorkStatus
|
WorkStatus = dtWorkStatus
|
||||||
WorkWarning = dtWorkWarning
|
WorkWarning = dtWorkWarning
|
||||||
|
@ -9,7 +9,6 @@ import (
|
|||||||
var (
|
var (
|
||||||
ErrWorkWarning = errors.New("Work warning")
|
ErrWorkWarning = errors.New("Work warning")
|
||||||
ErrInvalidData = errors.New("Invalid data")
|
ErrInvalidData = errors.New("Invalid data")
|
||||||
ErrInvalidId = errors.New("Invalid ID")
|
|
||||||
ErrWorkFail = errors.New("Work fail")
|
ErrWorkFail = errors.New("Work fail")
|
||||||
ErrWorkException = errors.New("Work exeption")
|
ErrWorkException = errors.New("Work exeption")
|
||||||
ErrDataType = errors.New("Invalid data type")
|
ErrDataType = errors.New("Invalid data type")
|
||||||
|
@ -32,7 +32,7 @@ func (ai *autoincId) Id() string {
|
|||||||
return strconv.FormatInt(next, 10)
|
return strconv.FormatInt(next, 10)
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewAutoIncId returns an autoincrement ID generator
|
// Return an autoincrement ID generator
|
||||||
func NewAutoIncId() IdGenerator {
|
func NewAutoIncId() IdGenerator {
|
||||||
// we'll consider the nano fraction of a second at startup unique
|
// we'll consider the nano fraction of a second at startup unique
|
||||||
// and count up from there.
|
// and count up from there.
|
||||||
|
@ -11,18 +11,20 @@ const (
|
|||||||
)
|
)
|
||||||
|
|
||||||
var (
|
var (
|
||||||
ErrNotFound = errors.New("Server Not Found")
|
ErrNotFound = errors.New("Server Not Found")
|
||||||
|
SelectWithRate = selectWithRate
|
||||||
|
SelectRandom = selectRandom
|
||||||
)
|
)
|
||||||
|
|
||||||
type PoolClient struct {
|
type poolClient struct {
|
||||||
*Client
|
*Client
|
||||||
Rate int
|
Rate int
|
||||||
mutex sync.Mutex
|
mutex sync.Mutex
|
||||||
}
|
}
|
||||||
|
|
||||||
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 {
|
||||||
@ -34,7 +36,7 @@ func SelectWithRate(pool map[string]*PoolClient,
|
|||||||
return last
|
return last
|
||||||
}
|
}
|
||||||
|
|
||||||
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
|
||||||
@ -50,17 +52,17 @@ func SelectRandom(pool map[string]*PoolClient,
|
|||||||
type Pool struct {
|
type Pool struct {
|
||||||
SelectionHandler SelectionHandler
|
SelectionHandler SelectionHandler
|
||||||
ErrorHandler ErrorHandler
|
ErrorHandler ErrorHandler
|
||||||
Clients map[string]*PoolClient
|
|
||||||
|
|
||||||
last string
|
clients map[string]*poolClient
|
||||||
|
last string
|
||||||
|
|
||||||
mutex sync.Mutex
|
mutex sync.Mutex
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewPool returns a new pool.
|
// Return a new pool.
|
||||||
func NewPool() (pool *Pool) {
|
func NewPool() (pool *Pool) {
|
||||||
return &Pool{
|
return &Pool{
|
||||||
Clients: make(map[string]*PoolClient, poolSize),
|
clients: make(map[string]*poolClient, poolSize),
|
||||||
SelectionHandler: SelectWithRate,
|
SelectionHandler: SelectWithRate,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -69,16 +71,16 @@ func NewPool() (pool *Pool) {
|
|||||||
func (pool *Pool) Add(net, 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
|
||||||
var ok bool
|
var ok bool
|
||||||
if item, ok = pool.Clients[addr]; ok {
|
if item, ok = pool.clients[addr]; ok {
|
||||||
item.Rate = rate
|
item.Rate = rate
|
||||||
} else {
|
} else {
|
||||||
var client *Client
|
var client *Client
|
||||||
client, err = New(net, addr)
|
client, err = New(net, addr)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
item = &PoolClient{Client: client, Rate: rate}
|
item = &poolClient{Client: client, Rate: rate}
|
||||||
pool.Clients[addr] = item
|
pool.clients[addr] = item
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return
|
return
|
||||||
@ -88,7 +90,7 @@ func (pool *Pool) Add(net, addr string, rate int) (err error) {
|
|||||||
func (pool *Pool) Remove(addr string) {
|
func (pool *Pool) Remove(addr string) {
|
||||||
pool.mutex.Lock()
|
pool.mutex.Lock()
|
||||||
defer pool.mutex.Unlock()
|
defer pool.mutex.Unlock()
|
||||||
delete(pool.Clients, addr)
|
delete(pool.clients, addr)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (pool *Pool) Do(funcname string, data []byte,
|
func (pool *Pool) Do(funcname string, data []byte,
|
||||||
@ -111,10 +113,10 @@ func (pool *Pool) DoBg(funcname string, data []byte,
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// Status gets job status from job server.
|
// Get job status from job server.
|
||||||
// !!!Not fully tested.!!!
|
// !!!Not fully tested.!!!
|
||||||
func (pool *Pool) Status(addr, handle string) (status *Status, err error) {
|
func (pool *Pool) Status(addr, handle string) (status *Status, err error) {
|
||||||
if client, ok := pool.Clients[addr]; ok {
|
if client, ok := pool.clients[addr]; ok {
|
||||||
client.Lock()
|
client.Lock()
|
||||||
defer client.Unlock()
|
defer client.Unlock()
|
||||||
status, err = client.Status(handle)
|
status, err = client.Status(handle)
|
||||||
@ -126,12 +128,12 @@ func (pool *Pool) Status(addr, handle string) (status *Status, err error) {
|
|||||||
|
|
||||||
// Send a something out, get the samething back.
|
// Send a something out, get the samething back.
|
||||||
func (pool *Pool) Echo(addr string, data []byte) (echo []byte, err error) {
|
func (pool *Pool) Echo(addr string, data []byte) (echo []byte, err error) {
|
||||||
var client *PoolClient
|
var client *poolClient
|
||||||
if addr == "" {
|
if addr == "" {
|
||||||
client = pool.selectServer()
|
client = pool.selectServer()
|
||||||
} else {
|
} else {
|
||||||
var ok bool
|
var ok bool
|
||||||
if client, ok = pool.Clients[addr]; !ok {
|
if client, ok = pool.clients[addr]; !ok {
|
||||||
err = ErrNotFound
|
err = ErrNotFound
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@ -145,18 +147,18 @@ func (pool *Pool) Echo(addr string, data []byte) (echo []byte, err error) {
|
|||||||
// Close
|
// Close
|
||||||
func (pool *Pool) Close() (err map[string]error) {
|
func (pool *Pool) Close() (err map[string]error) {
|
||||||
err = make(map[string]error)
|
err = make(map[string]error)
|
||||||
for _, c := range pool.Clients {
|
for _, c := range pool.clients {
|
||||||
err[c.addr] = c.Close()
|
err[c.addr] = c.Close()
|
||||||
}
|
}
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// selecting server
|
// 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)
|
||||||
var ok bool
|
var ok bool
|
||||||
if client, ok = pool.Clients[addr]; ok {
|
if client, ok = pool.clients[addr]; ok {
|
||||||
pool.last = addr
|
pool.last = addr
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
|
@ -9,9 +9,6 @@ var (
|
|||||||
)
|
)
|
||||||
|
|
||||||
func TestPoolAdd(t *testing.T) {
|
func TestPoolAdd(t *testing.T) {
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
t.Log("Add servers")
|
t.Log("Add servers")
|
||||||
c := 2
|
c := 2
|
||||||
if err := pool.Add("tcp4", "127.0.0.1:4730", 1); err != nil {
|
if err := pool.Add("tcp4", "127.0.0.1:4730", 1); err != nil {
|
||||||
@ -21,15 +18,12 @@ func TestPoolAdd(t *testing.T) {
|
|||||||
t.Log(err)
|
t.Log(err)
|
||||||
c -= 1
|
c -= 1
|
||||||
}
|
}
|
||||||
if len(pool.Clients) != c {
|
if len(pool.clients) != c {
|
||||||
t.Errorf("%d servers expected, %d got.", c, len(pool.Clients))
|
t.Errorf("%d servers expected, %d got.", c, len(pool.clients))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestPoolEcho(t *testing.T) {
|
func TestPoolEcho(t *testing.T) {
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
echo, err := pool.Echo("", []byte(TestStr))
|
echo, err := pool.Echo("", []byte(TestStr))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Error(err)
|
t.Error(err)
|
||||||
@ -47,9 +41,6 @@ func TestPoolEcho(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestPoolDoBg(t *testing.T) {
|
func TestPoolDoBg(t *testing.T) {
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
addr, handle, err := pool.DoBg("ToUpper",
|
addr, handle, err := pool.DoBg("ToUpper",
|
||||||
[]byte("abcdef"), JobLow)
|
[]byte("abcdef"), JobLow)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@ -64,9 +55,6 @@ func TestPoolDoBg(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestPoolDo(t *testing.T) {
|
func TestPoolDo(t *testing.T) {
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
jobHandler := func(job *Response) {
|
jobHandler := func(job *Response) {
|
||||||
str := string(job.Data)
|
str := string(job.Data)
|
||||||
if str == "ABCDEF" {
|
if str == "ABCDEF" {
|
||||||
@ -89,9 +77,6 @@ func TestPoolDo(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestPoolStatus(t *testing.T) {
|
func TestPoolStatus(t *testing.T) {
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
status, err := pool.Status("127.0.0.1:4730", "handle not exists")
|
status, err := pool.Status("127.0.0.1:4730", "handle not exists")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Error(err)
|
t.Error(err)
|
||||||
@ -129,9 +114,6 @@ func TestPoolStatus(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestPoolClose(t *testing.T) {
|
func TestPoolClose(t *testing.T) {
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
return
|
return
|
||||||
if err := pool.Close(); err != nil {
|
if err := pool.Close(); err != nil {
|
||||||
t.Error(err)
|
t.Error(err)
|
||||||
|
@ -76,7 +76,7 @@ func decodeResponse(data []byte) (resp *Response, l int, err error) {
|
|||||||
case dtJobCreated:
|
case dtJobCreated:
|
||||||
resp.Handle = string(dt)
|
resp.Handle = string(dt)
|
||||||
case dtStatusRes, dtWorkData, dtWorkWarning, dtWorkStatus,
|
case dtStatusRes, dtWorkData, dtWorkWarning, dtWorkStatus,
|
||||||
dtWorkComplete, dtWorkException:
|
dtWorkComplete, dtWorkFail, dtWorkException:
|
||||||
s := bytes.SplitN(dt, []byte{'\x00'}, 2)
|
s := bytes.SplitN(dt, []byte{'\x00'}, 2)
|
||||||
if len(s) >= 2 {
|
if len(s) >= 2 {
|
||||||
resp.Handle = string(s[0])
|
resp.Handle = string(s[0])
|
||||||
@ -85,14 +85,6 @@ func decodeResponse(data []byte) (resp *Response, l int, err error) {
|
|||||||
err = fmt.Errorf("Invalid data: %v", data)
|
err = fmt.Errorf("Invalid data: %v", data)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
case dtWorkFail:
|
|
||||||
s := bytes.SplitN(dt, []byte{'\x00'}, 2)
|
|
||||||
if len(s) >= 1 {
|
|
||||||
resp.Handle = string(s[0])
|
|
||||||
} else {
|
|
||||||
err = fmt.Errorf("Invalid data: %v", data)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
case dtEchoRes:
|
case dtEchoRes:
|
||||||
fallthrough
|
fallthrough
|
||||||
default:
|
default:
|
||||||
|
@ -3,8 +3,8 @@ package main
|
|||||||
import (
|
import (
|
||||||
"github.com/mikespook/gearman-go/client"
|
"github.com/mikespook/gearman-go/client"
|
||||||
"log"
|
"log"
|
||||||
"os"
|
|
||||||
"sync"
|
"sync"
|
||||||
|
"os"
|
||||||
)
|
)
|
||||||
|
|
||||||
func main() {
|
func main() {
|
||||||
|
@ -1,33 +0,0 @@
|
|||||||
#!/usr/bin/perl
|
|
||||||
|
|
||||||
# Runs 20 children that expose "PerlToUpper" before returning the result.
|
|
||||||
|
|
||||||
use strict; use warnings;
|
|
||||||
use constant CHILDREN => 20;
|
|
||||||
use Time::HiRes qw(usleep);
|
|
||||||
use Gearman::Worker;
|
|
||||||
|
|
||||||
$|++;
|
|
||||||
my @child_pids;
|
|
||||||
for (1 .. CHILDREN) {
|
|
||||||
if (my $pid = fork) {
|
|
||||||
push @child_pids, $pid;
|
|
||||||
next;
|
|
||||||
}
|
|
||||||
eval {
|
|
||||||
my $w = Gearman::Worker->new(job_servers => '127.0.0.1:4730');
|
|
||||||
$w->register_function(PerlToUpper => sub { print "."; uc $_[0]->arg });
|
|
||||||
$w->work while 1;
|
|
||||||
};
|
|
||||||
warn $@ if $@;
|
|
||||||
exit 0;
|
|
||||||
}
|
|
||||||
|
|
||||||
$SIG{INT} = $SIG{HUP} = sub {
|
|
||||||
kill 9, @child_pids;
|
|
||||||
print "\nChildren shut down, gracefully exiting\n";
|
|
||||||
exit 0;
|
|
||||||
};
|
|
||||||
|
|
||||||
printf "Forked %d children, serving 'PerlToUpper' function to gearman\n", CHILDREN;
|
|
||||||
sleep;
|
|
@ -1,14 +1,13 @@
|
|||||||
package main
|
package main
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"github.com/mikespook/gearman-go/worker"
|
||||||
|
"github.com/mikespook/golib/signal"
|
||||||
"log"
|
"log"
|
||||||
"net"
|
|
||||||
"os"
|
"os"
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
"net"
|
||||||
"github.com/mikespook/gearman-go/worker"
|
|
||||||
"github.com/mikespook/golib/signal"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
func ToUpper(job worker.Job) ([]byte, error) {
|
func ToUpper(job worker.Job) ([]byte, error) {
|
||||||
@ -42,7 +41,7 @@ func main() {
|
|||||||
w.ErrorHandler = func(e error) {
|
w.ErrorHandler = func(e error) {
|
||||||
log.Println(e)
|
log.Println(e)
|
||||||
if opErr, ok := e.(*net.OpError); ok {
|
if opErr, ok := e.(*net.OpError); ok {
|
||||||
if !opErr.Temporary() {
|
if ! opErr.Temporary() {
|
||||||
proc, err := os.FindProcess(os.Getpid())
|
proc, err := os.FindProcess(os.Getpid())
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Println(err)
|
log.Println(err)
|
||||||
@ -69,6 +68,7 @@ func main() {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
go w.Work()
|
go w.Work()
|
||||||
signal.Bind(os.Interrupt, func() uint { return signal.BreakExit })
|
sh := signal.NewHandler()
|
||||||
signal.Wait()
|
sh.Bind(os.Interrupt, func() bool { return true })
|
||||||
|
sh.Loop()
|
||||||
}
|
}
|
||||||
|
@ -17,3 +17,8 @@ in an easy way.
|
|||||||
import "github.com/mikespook/gearman-go/worker"
|
import "github.com/mikespook/gearman-go/worker"
|
||||||
*/
|
*/
|
||||||
package gearman
|
package gearman
|
||||||
|
|
||||||
|
import (
|
||||||
|
_ "github.com/mikespook/gearman-go/client"
|
||||||
|
_ "github.com/mikespook/gearman-go/worker"
|
||||||
|
)
|
||||||
|
126
worker/agent.go
126
worker/agent.go
@ -2,9 +2,6 @@ package worker
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"bufio"
|
"bufio"
|
||||||
"bytes"
|
|
||||||
"encoding/binary"
|
|
||||||
"io"
|
|
||||||
"net"
|
"net"
|
||||||
"sync"
|
"sync"
|
||||||
)
|
)
|
||||||
@ -49,26 +46,20 @@ func (a *agent) work() {
|
|||||||
a.worker.err(err.(error))
|
a.worker.err(err.(error))
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
var inpack *inPack
|
var inpack *inPack
|
||||||
var l int
|
var l int
|
||||||
var err error
|
var err error
|
||||||
var data, leftdata []byte
|
var data, leftdata []byte
|
||||||
for {
|
for {
|
||||||
if data, err = a.read(); err != nil {
|
if data, err = a.read(bufferSize); err != nil {
|
||||||
if opErr, ok := err.(*net.OpError); ok {
|
if opErr, ok := err.(*net.OpError); ok {
|
||||||
if opErr.Temporary() {
|
if opErr.Temporary() {
|
||||||
continue
|
continue
|
||||||
} else {
|
}else{
|
||||||
a.disconnect_error(err)
|
a.worker.err(err)
|
||||||
// else - we're probably dc'ing due to a Close()
|
|
||||||
|
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
|
|
||||||
} else if err == io.EOF {
|
|
||||||
a.disconnect_error(err)
|
|
||||||
break
|
|
||||||
}
|
}
|
||||||
a.worker.err(err)
|
a.worker.err(err)
|
||||||
// If it is unexpected error and the connection wasn't
|
// If it is unexpected error and the connection wasn't
|
||||||
@ -90,41 +81,17 @@ func (a *agent) work() {
|
|||||||
leftdata = data
|
leftdata = data
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
for {
|
if inpack, l, err = decodeInPack(data); err != nil {
|
||||||
if inpack, l, err = decodeInPack(data); err != nil {
|
a.worker.err(err)
|
||||||
a.worker.err(err)
|
leftdata = data
|
||||||
leftdata = data
|
continue
|
||||||
break
|
|
||||||
} else {
|
|
||||||
leftdata = nil
|
|
||||||
inpack.a = a
|
|
||||||
select {
|
|
||||||
case <-a.worker.closed:
|
|
||||||
return
|
|
||||||
default:
|
|
||||||
}
|
|
||||||
a.worker.in <- inpack
|
|
||||||
if len(data) == l {
|
|
||||||
break
|
|
||||||
}
|
|
||||||
if len(data) > l {
|
|
||||||
data = data[l:]
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
leftdata = nil
|
||||||
}
|
inpack.a = a
|
||||||
|
a.worker.in <- inpack
|
||||||
func (a *agent) disconnect_error(err error) {
|
if len(data) > l {
|
||||||
a.Lock()
|
leftdata = data[l:]
|
||||||
defer a.Unlock()
|
|
||||||
|
|
||||||
if a.conn != nil {
|
|
||||||
err = &WorkerDisconnectError{
|
|
||||||
err: err,
|
|
||||||
agent: a,
|
|
||||||
}
|
}
|
||||||
a.worker.err(err)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -140,17 +107,9 @@ func (a *agent) Close() {
|
|||||||
func (a *agent) Grab() {
|
func (a *agent) Grab() {
|
||||||
a.Lock()
|
a.Lock()
|
||||||
defer a.Unlock()
|
defer a.Unlock()
|
||||||
a.grab()
|
|
||||||
}
|
|
||||||
|
|
||||||
func (a *agent) grab() bool {
|
|
||||||
if a.worker.closed != nil {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
outpack := getOutPack()
|
outpack := getOutPack()
|
||||||
outpack.dataType = dtGrabJobUniq
|
outpack.dataType = dtGrabJobUniq
|
||||||
a.write(outpack)
|
a.write(outpack)
|
||||||
return true
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (a *agent) PreSleep() {
|
func (a *agent) PreSleep() {
|
||||||
@ -161,51 +120,21 @@ func (a *agent) PreSleep() {
|
|||||||
a.write(outpack)
|
a.write(outpack)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (a *agent) reconnect() error {
|
|
||||||
a.Lock()
|
|
||||||
defer a.Unlock()
|
|
||||||
conn, err := net.Dial(a.net, a.addr)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
a.conn = conn
|
|
||||||
a.rw = bufio.NewReadWriter(bufio.NewReader(a.conn),
|
|
||||||
bufio.NewWriter(a.conn))
|
|
||||||
|
|
||||||
a.worker.reRegisterFuncsForAgent(a)
|
|
||||||
if a.grab() {
|
|
||||||
go a.work()
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// read length bytes from the socket
|
// read length bytes from the socket
|
||||||
func (a *agent) read() (data []byte, err error) {
|
func (a *agent) read(length int) (data []byte, err error) {
|
||||||
n := 0
|
n := 0
|
||||||
|
buf := getBuffer(bufferSize)
|
||||||
tmp := getBuffer(bufferSize)
|
// read until data can be unpacked
|
||||||
var buf bytes.Buffer
|
for i := length; i > 0 || len(data) < minPacketLength; i -= n {
|
||||||
|
if n, err = a.rw.Read(buf); err != nil {
|
||||||
// read the header so we can get the length of the data
|
return
|
||||||
if n, err = a.rw.Read(tmp); err != nil {
|
}
|
||||||
return
|
data = append(data, buf[0:n]...)
|
||||||
}
|
if n < bufferSize {
|
||||||
dl := int(binary.BigEndian.Uint32(tmp[8:12]))
|
break
|
||||||
|
|
||||||
// write what we read so far
|
|
||||||
buf.Write(tmp[:n])
|
|
||||||
|
|
||||||
// read until we receive all the data
|
|
||||||
for buf.Len() < dl+minPacketLength {
|
|
||||||
if n, err = a.rw.Read(tmp); err != nil {
|
|
||||||
return buf.Bytes(), err
|
|
||||||
}
|
}
|
||||||
|
|
||||||
buf.Write(tmp[:n])
|
|
||||||
}
|
}
|
||||||
|
return
|
||||||
return buf.Bytes(), err
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Internal write the encoded job.
|
// Internal write the encoded job.
|
||||||
@ -220,10 +149,3 @@ func (a *agent) write(outpack *outPack) (err error) {
|
|||||||
}
|
}
|
||||||
return a.rw.Flush()
|
return a.rw.Flush()
|
||||||
}
|
}
|
||||||
|
|
||||||
// Write with lock
|
|
||||||
func (a *agent) Write(outpack *outPack) (err error) {
|
|
||||||
a.Lock()
|
|
||||||
defer a.Unlock()
|
|
||||||
return a.write(outpack)
|
|
||||||
}
|
|
||||||
|
@ -53,4 +53,5 @@ func ExampleWorker() {
|
|||||||
w.Echo([]byte("Hello"))
|
w.Echo([]byte("Hello"))
|
||||||
// Waiting results
|
// Waiting results
|
||||||
wg.Wait()
|
wg.Wait()
|
||||||
|
// Output: Hello
|
||||||
}
|
}
|
||||||
|
175
worker/worker.go
175
worker/worker.go
@ -3,8 +3,8 @@
|
|||||||
package worker
|
package worker
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"encoding/binary"
|
||||||
"fmt"
|
"fmt"
|
||||||
"strconv"
|
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
)
|
)
|
||||||
@ -18,22 +18,19 @@ const (
|
|||||||
// It can connect to multi-server and grab jobs.
|
// It can connect to multi-server and grab jobs.
|
||||||
type Worker struct {
|
type Worker struct {
|
||||||
sync.Mutex
|
sync.Mutex
|
||||||
agents []*agent
|
agents []*agent
|
||||||
funcs jobFuncs
|
funcs jobFuncs
|
||||||
in chan *inPack
|
in chan *inPack
|
||||||
running bool
|
running bool
|
||||||
ready bool
|
ready bool
|
||||||
jobLeftNum int64
|
|
||||||
|
|
||||||
Id string
|
Id string
|
||||||
ErrorHandler ErrorHandler
|
ErrorHandler ErrorHandler
|
||||||
JobHandler JobHandler
|
JobHandler JobHandler
|
||||||
limit chan bool
|
limit chan bool
|
||||||
closed chan struct{}
|
|
||||||
leftJobs chan struct{}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// New returns a worker.
|
// Return a worker.
|
||||||
//
|
//
|
||||||
// If limit is set to Unlimited(=0), the worker will grab all jobs
|
// If limit is set to Unlimited(=0), the worker will grab all jobs
|
||||||
// and execute them parallelly.
|
// and execute them parallelly.
|
||||||
@ -59,7 +56,7 @@ func (worker *Worker) err(e error) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// AddServer adds a Gearman job server.
|
// Add a Gearman job server.
|
||||||
//
|
//
|
||||||
// addr should be formated as 'host:port'.
|
// addr should be formated as 'host:port'.
|
||||||
func (worker *Worker) AddServer(net, addr string) (err error) {
|
func (worker *Worker) AddServer(net, addr string) (err error) {
|
||||||
@ -75,11 +72,11 @@ func (worker *Worker) AddServer(net, addr string) (err error) {
|
|||||||
// Broadcast an outpack to all Gearman server.
|
// Broadcast an outpack to all Gearman server.
|
||||||
func (worker *Worker) broadcast(outpack *outPack) {
|
func (worker *Worker) broadcast(outpack *outPack) {
|
||||||
for _, v := range worker.agents {
|
for _, v := range worker.agents {
|
||||||
v.Write(outpack)
|
v.write(outpack)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// AddFunc adds a function.
|
// Add a function.
|
||||||
// Set timeout as Unlimited(=0) to disable executing timeout.
|
// Set timeout as Unlimited(=0) to disable executing timeout.
|
||||||
func (worker *Worker) AddFunc(funcname string,
|
func (worker *Worker) AddFunc(funcname string,
|
||||||
f JobFunc, timeout uint32) (err error) {
|
f JobFunc, timeout uint32) (err error) {
|
||||||
@ -97,11 +94,6 @@ func (worker *Worker) AddFunc(funcname string,
|
|||||||
|
|
||||||
// inner add
|
// inner add
|
||||||
func (worker *Worker) addFunc(funcname string, timeout uint32) {
|
func (worker *Worker) addFunc(funcname string, timeout uint32) {
|
||||||
outpack := prepFuncOutpack(funcname, timeout)
|
|
||||||
worker.broadcast(outpack)
|
|
||||||
}
|
|
||||||
|
|
||||||
func prepFuncOutpack(funcname string, timeout uint32) *outPack {
|
|
||||||
outpack := getOutPack()
|
outpack := getOutPack()
|
||||||
if timeout == 0 {
|
if timeout == 0 {
|
||||||
outpack.dataType = dtCanDo
|
outpack.dataType = dtCanDo
|
||||||
@ -109,17 +101,15 @@ func prepFuncOutpack(funcname string, timeout uint32) *outPack {
|
|||||||
} else {
|
} else {
|
||||||
outpack.dataType = dtCanDoTimeout
|
outpack.dataType = dtCanDoTimeout
|
||||||
l := len(funcname)
|
l := len(funcname)
|
||||||
|
outpack.data = getBuffer(l + 5)
|
||||||
timeoutString := strconv.FormatUint(uint64(timeout), 10)
|
|
||||||
outpack.data = getBuffer(l + len(timeoutString) + 1)
|
|
||||||
copy(outpack.data, []byte(funcname))
|
copy(outpack.data, []byte(funcname))
|
||||||
outpack.data[l] = '\x00'
|
outpack.data[l] = '\x00'
|
||||||
copy(outpack.data[l+1:], []byte(timeoutString))
|
binary.BigEndian.PutUint32(outpack.data[l+1:], timeout)
|
||||||
}
|
}
|
||||||
return outpack
|
worker.broadcast(outpack)
|
||||||
}
|
}
|
||||||
|
|
||||||
// RemoveFunc removes a function.
|
// Remove a function.
|
||||||
func (worker *Worker) RemoveFunc(funcname string) (err error) {
|
func (worker *Worker) RemoveFunc(funcname string) (err error) {
|
||||||
worker.Lock()
|
worker.Lock()
|
||||||
defer worker.Unlock()
|
defer worker.Unlock()
|
||||||
@ -150,20 +140,14 @@ func (worker *Worker) handleInPack(inpack *inPack) {
|
|||||||
inpack.a.Grab()
|
inpack.a.Grab()
|
||||||
case dtJobAssign, dtJobAssignUniq:
|
case dtJobAssign, dtJobAssignUniq:
|
||||||
go func() {
|
go func() {
|
||||||
go func() {
|
if err := worker.exec(inpack); err != nil {
|
||||||
worker.incrExecJobNum()
|
worker.err(err)
|
||||||
defer func() {
|
|
||||||
worker.decrExecJobNum()
|
|
||||||
}()
|
|
||||||
if err := worker.exec(inpack); err != nil {
|
|
||||||
worker.err(err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
if worker.limit != nil {
|
|
||||||
worker.limit <- true
|
|
||||||
}
|
}
|
||||||
inpack.a.Grab()
|
|
||||||
}()
|
}()
|
||||||
|
if worker.limit != nil {
|
||||||
|
worker.limit <- true
|
||||||
|
}
|
||||||
|
inpack.a.Grab()
|
||||||
case dtError:
|
case dtError:
|
||||||
worker.err(inpack.Err())
|
worker.err(inpack.Err())
|
||||||
fallthrough
|
fallthrough
|
||||||
@ -195,40 +179,30 @@ func (worker *Worker) Ready() (err error) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// Work start main loop (blocking)
|
// Main loop, block here
|
||||||
// Most of time, this should be evaluated in goroutine.
|
// Most of time, this should be evaluated in goroutine.
|
||||||
func (worker *Worker) Work() {
|
func (worker *Worker) Work() {
|
||||||
if !worker.ready {
|
if ! worker.ready {
|
||||||
// didn't run Ready beforehand, so we'll have to do it:
|
// didn't run Ready beforehand, so we'll have to do it:
|
||||||
err := worker.Ready()
|
err := worker.Ready()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
panic(err)
|
panic( err )
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
worker.Lock()
|
defer func() {
|
||||||
|
for _, a := range worker.agents {
|
||||||
|
a.Close()
|
||||||
|
}
|
||||||
|
}()
|
||||||
worker.running = true
|
worker.running = true
|
||||||
worker.Unlock()
|
|
||||||
|
|
||||||
for _, a := range worker.agents {
|
for _, a := range worker.agents {
|
||||||
a.Grab()
|
a.Grab()
|
||||||
}
|
}
|
||||||
// 执行任务(阻塞)
|
|
||||||
var inpack *inPack
|
var inpack *inPack
|
||||||
for inpack = range worker.in {
|
for inpack = range worker.in {
|
||||||
worker.handleInPack(inpack)
|
worker.handleInPack(inpack)
|
||||||
}
|
}
|
||||||
// 关闭Worker进程后 等待任务完成后退出
|
|
||||||
worker.Lock()
|
|
||||||
leftJobNum := int(worker.jobLeftNum)
|
|
||||||
worker.Unlock()
|
|
||||||
if worker.leftJobs != nil {
|
|
||||||
for i := 0; i < leftJobNum; i++ {
|
|
||||||
<-worker.leftJobs
|
|
||||||
}
|
|
||||||
}
|
|
||||||
worker.Reset()
|
|
||||||
worker.close()
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// custome handling warper
|
// custome handling warper
|
||||||
@ -243,22 +217,10 @@ func (worker *Worker) customeHandler(inpack *inPack) {
|
|||||||
// Close connection and exit main loop
|
// Close connection and exit main loop
|
||||||
func (worker *Worker) Close() {
|
func (worker *Worker) Close() {
|
||||||
worker.Lock()
|
worker.Lock()
|
||||||
defer worker.Unlock()
|
worker.Unlock()
|
||||||
if worker.running == true && worker.closed == nil {
|
if worker.running == true {
|
||||||
worker.closed = make(chan struct{}, 1)
|
|
||||||
worker.closed <- struct{}{}
|
|
||||||
worker.running = false
|
worker.running = false
|
||||||
close(worker.in)
|
close(worker.in)
|
||||||
// 创建关闭后执行中的任务列表
|
|
||||||
if worker.jobLeftNum != 0 {
|
|
||||||
worker.leftJobs = make(chan struct{}, worker.jobLeftNum+int64(len(worker.in)))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func (worker *Worker) close() {
|
|
||||||
for _, a := range worker.agents {
|
|
||||||
a.Close()
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -270,7 +232,7 @@ func (worker *Worker) Echo(data []byte) {
|
|||||||
worker.broadcast(outpack)
|
worker.broadcast(outpack)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Reset removes all of functions.
|
// Remove all of functions.
|
||||||
// Both from the worker and job servers.
|
// Both from the worker and job servers.
|
||||||
func (worker *Worker) Reset() {
|
func (worker *Worker) Reset() {
|
||||||
outpack := getOutPack()
|
outpack := getOutPack()
|
||||||
@ -288,23 +250,6 @@ func (worker *Worker) SetId(id string) {
|
|||||||
worker.broadcast(outpack)
|
worker.broadcast(outpack)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (worker *Worker) incrExecJobNum() int64 {
|
|
||||||
worker.Lock()
|
|
||||||
defer worker.Unlock()
|
|
||||||
worker.jobLeftNum++
|
|
||||||
return worker.jobLeftNum
|
|
||||||
}
|
|
||||||
|
|
||||||
func (worker *Worker) decrExecJobNum() int64 {
|
|
||||||
worker.Lock()
|
|
||||||
defer worker.Unlock()
|
|
||||||
worker.jobLeftNum--
|
|
||||||
if worker.jobLeftNum < 0 {
|
|
||||||
worker.jobLeftNum = 0
|
|
||||||
}
|
|
||||||
return worker.jobLeftNum
|
|
||||||
}
|
|
||||||
|
|
||||||
// inner job executing
|
// inner job executing
|
||||||
func (worker *Worker) exec(inpack *inPack) (err error) {
|
func (worker *Worker) exec(inpack *inPack) (err error) {
|
||||||
defer func() {
|
defer func() {
|
||||||
@ -330,36 +275,24 @@ func (worker *Worker) exec(inpack *inPack) (err error) {
|
|||||||
} else {
|
} else {
|
||||||
r = execTimeout(f.f, inpack, time.Duration(f.timeout)*time.Second)
|
r = execTimeout(f.f, inpack, time.Duration(f.timeout)*time.Second)
|
||||||
}
|
}
|
||||||
//if worker.running {
|
if worker.running {
|
||||||
outpack := getOutPack()
|
outpack := getOutPack()
|
||||||
if r.err == nil {
|
if r.err == nil {
|
||||||
outpack.dataType = dtWorkComplete
|
outpack.dataType = dtWorkComplete
|
||||||
} else {
|
|
||||||
if len(r.data) == 0 {
|
|
||||||
outpack.dataType = dtWorkFail
|
|
||||||
} else {
|
} else {
|
||||||
outpack.dataType = dtWorkException
|
if len(r.data) == 0 {
|
||||||
|
outpack.dataType = dtWorkFail
|
||||||
|
} else {
|
||||||
|
outpack.dataType = dtWorkException
|
||||||
|
}
|
||||||
|
err = r.err
|
||||||
}
|
}
|
||||||
err = r.err
|
outpack.handle = inpack.handle
|
||||||
|
outpack.data = r.data
|
||||||
|
inpack.a.write(outpack)
|
||||||
}
|
}
|
||||||
outpack.handle = inpack.handle
|
|
||||||
outpack.data = r.data
|
|
||||||
_ = inpack.a.Write(outpack)
|
|
||||||
if worker.leftJobs != nil {
|
|
||||||
worker.leftJobs <- struct{}{}
|
|
||||||
}
|
|
||||||
//}
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
func (worker *Worker) reRegisterFuncsForAgent(a *agent) {
|
|
||||||
worker.Lock()
|
|
||||||
defer worker.Unlock()
|
|
||||||
for funcname, f := range worker.funcs {
|
|
||||||
outpack := prepFuncOutpack(funcname, f.timeout)
|
|
||||||
a.write(outpack)
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
||||||
// inner result
|
// inner result
|
||||||
type result struct {
|
type result struct {
|
||||||
@ -383,23 +316,3 @@ func execTimeout(f JobFunc, job Job, timeout time.Duration) (r *result) {
|
|||||||
}
|
}
|
||||||
return r
|
return r
|
||||||
}
|
}
|
||||||
|
|
||||||
// Error type passed when a worker connection disconnects
|
|
||||||
type WorkerDisconnectError struct {
|
|
||||||
err error
|
|
||||||
agent *agent
|
|
||||||
}
|
|
||||||
|
|
||||||
func (e *WorkerDisconnectError) Error() string {
|
|
||||||
return e.err.Error()
|
|
||||||
}
|
|
||||||
|
|
||||||
// Responds to the error by asking the worker to reconnect
|
|
||||||
func (e *WorkerDisconnectError) Reconnect() (err error) {
|
|
||||||
return e.agent.reconnect()
|
|
||||||
}
|
|
||||||
|
|
||||||
// Which server was this for?
|
|
||||||
func (e *WorkerDisconnectError) Server() (net string, addr string) {
|
|
||||||
return e.agent.net, e.agent.addr
|
|
||||||
}
|
|
||||||
|
170
worker/worker_disconnect_test.go
Normal file
170
worker/worker_disconnect_test.go
Normal file
@ -0,0 +1,170 @@
|
|||||||
|
package worker
|
||||||
|
|
||||||
|
import (
|
||||||
|
"../client"
|
||||||
|
"log"
|
||||||
|
"net"
|
||||||
|
"os/exec"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
const port = `3700`
|
||||||
|
|
||||||
|
var gearman_ready chan bool
|
||||||
|
var kill_gearman chan bool
|
||||||
|
var bye chan bool
|
||||||
|
|
||||||
|
func init() {
|
||||||
|
|
||||||
|
if check_gearman_present() {
|
||||||
|
panic(`Something already listening on our testing port. Chickening out of testing with it!`)
|
||||||
|
}
|
||||||
|
gearman_ready = make( chan bool )
|
||||||
|
kill_gearman = make( chan bool )
|
||||||
|
// TODO: verify port is clear
|
||||||
|
go run_gearman()
|
||||||
|
}
|
||||||
|
|
||||||
|
func run_gearman() {
|
||||||
|
gm_cmd := exec.Command(`/usr/sbin/gearmand`, `--port`, port)
|
||||||
|
start_err := gm_cmd.Start()
|
||||||
|
|
||||||
|
if start_err != nil {
|
||||||
|
panic(`could not start gearman, aborting test :` + start_err.Error())
|
||||||
|
}
|
||||||
|
|
||||||
|
// Make sure we clear up our gearman:
|
||||||
|
defer func() {
|
||||||
|
log.Println("killing gearmand")
|
||||||
|
gm_cmd.Process.Kill()
|
||||||
|
}()
|
||||||
|
|
||||||
|
for tries := 10; tries > 0; tries-- {
|
||||||
|
if check_gearman_present() {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
time.Sleep(250 * time.Millisecond)
|
||||||
|
}
|
||||||
|
|
||||||
|
if !check_gearman_present() {
|
||||||
|
panic(`Unable to start gearman aborting test`)
|
||||||
|
}
|
||||||
|
gearman_ready <- true
|
||||||
|
|
||||||
|
<- kill_gearman
|
||||||
|
}
|
||||||
|
|
||||||
|
func check_gearman_present() bool {
|
||||||
|
con, err := net.Dial(`tcp`, `127.0.0.1:`+port)
|
||||||
|
if err != nil {
|
||||||
|
log.Println("gearman not ready " + err.Error())
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
log.Println("gearman ready")
|
||||||
|
con.Close()
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
|
func check_gearman_is_dead() bool {
|
||||||
|
|
||||||
|
for tries := 10; tries > 0; tries-- {
|
||||||
|
if !check_gearman_present() {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
time.Sleep(250 * time.Millisecond)
|
||||||
|
}
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
/*
|
||||||
|
Checks for a disconnect whilst not working
|
||||||
|
*/
|
||||||
|
func TestBasicDisconnect(t *testing.T) {
|
||||||
|
<- gearman_ready
|
||||||
|
|
||||||
|
worker := New(Unlimited)
|
||||||
|
timeout := make(chan bool, 1)
|
||||||
|
done := make( chan bool, 1)
|
||||||
|
|
||||||
|
if err := worker.AddServer(Network, "127.0.0.1:" + port); err != nil {
|
||||||
|
t.Error(err)
|
||||||
|
}
|
||||||
|
work_done := false;
|
||||||
|
if err := worker.AddFunc("gearman-go-workertest",
|
||||||
|
func(j Job)(b []byte, e error){
|
||||||
|
work_done = true;
|
||||||
|
done <- true
|
||||||
|
return}, 0);
|
||||||
|
err != nil {
|
||||||
|
t.Error(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
worker.JobHandler = func( j Job ) error {
|
||||||
|
if( ! worker.ready ){
|
||||||
|
t.Error("Worker not ready as expected");
|
||||||
|
}
|
||||||
|
done <-true
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
handled_errors := false
|
||||||
|
|
||||||
|
c_error := make( chan bool)
|
||||||
|
worker.ErrorHandler = func( e error ){
|
||||||
|
log.Println( e )
|
||||||
|
handled_errors = true
|
||||||
|
c_error <- true
|
||||||
|
}
|
||||||
|
|
||||||
|
go func() {
|
||||||
|
time.Sleep(5 * time.Second)
|
||||||
|
timeout <- true
|
||||||
|
}()
|
||||||
|
|
||||||
|
err := worker.Ready()
|
||||||
|
|
||||||
|
if err != nil {
|
||||||
|
t.Error(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
go worker.Work()
|
||||||
|
|
||||||
|
kill_gearman <- true
|
||||||
|
|
||||||
|
check_gearman_is_dead()
|
||||||
|
go run_gearman()
|
||||||
|
|
||||||
|
select {
|
||||||
|
case <-gearman_ready:
|
||||||
|
case <-timeout:
|
||||||
|
}
|
||||||
|
|
||||||
|
send_client_request()
|
||||||
|
|
||||||
|
select {
|
||||||
|
case <- done:
|
||||||
|
t.Error("Client request handled (somehow), did we magically reconnect?")
|
||||||
|
case <-timeout:
|
||||||
|
t.Error("Test timed out waiting for the error handler")
|
||||||
|
case <-c_error:
|
||||||
|
// error was handled!
|
||||||
|
}
|
||||||
|
|
||||||
|
kill_gearman <- true
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
func send_client_request(){
|
||||||
|
log.Println("sending client request");
|
||||||
|
c, err := client.New( Network, "127.0.0.1:" + port )
|
||||||
|
if err == nil {
|
||||||
|
_, err = c.DoBg("gearman-go-workertest", []byte{}, client.JobHigh)
|
||||||
|
if err != nil {
|
||||||
|
log.Println( "error sending client request " + err.Error() )
|
||||||
|
}
|
||||||
|
|
||||||
|
}else{
|
||||||
|
log.Println( "error with client " + err.Error() )
|
||||||
|
}
|
||||||
|
}
|
@ -1,59 +0,0 @@
|
|||||||
package worker
|
|
||||||
|
|
||||||
import (
|
|
||||||
"fmt"
|
|
||||||
"sync"
|
|
||||||
"testing"
|
|
||||||
)
|
|
||||||
|
|
||||||
func TestWorkerRace(t *testing.T) {
|
|
||||||
// from example worker
|
|
||||||
// An example of worker
|
|
||||||
w := New(Unlimited)
|
|
||||||
defer w.Close()
|
|
||||||
// Add a gearman job server
|
|
||||||
if err := w.AddServer(Network, "127.0.0.1:4730"); err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
// A function for handling jobs
|
|
||||||
foobar := func(job Job) ([]byte, error) {
|
|
||||||
// Do nothing here
|
|
||||||
return nil, nil
|
|
||||||
}
|
|
||||||
// Add the function to worker
|
|
||||||
if err := w.AddFunc("foobar", foobar, 0); err != nil {
|
|
||||||
fmt.Println(err)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
var wg sync.WaitGroup
|
|
||||||
// A custome handler, for handling other results, eg. ECHO, dtError.
|
|
||||||
w.JobHandler = func(job Job) error {
|
|
||||||
if job.Err() == nil {
|
|
||||||
fmt.Println(string(job.Data()))
|
|
||||||
} else {
|
|
||||||
fmt.Println(job.Err())
|
|
||||||
}
|
|
||||||
wg.Done()
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
// An error handler for handling worker's internal errors.
|
|
||||||
w.ErrorHandler = func(e error) {
|
|
||||||
fmt.Println(e)
|
|
||||||
// Ignore the error or shutdown the worker
|
|
||||||
}
|
|
||||||
// Tell Gearman job server: I'm ready!
|
|
||||||
if err := w.Ready(); err != nil {
|
|
||||||
fmt.Println(err)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
// Running main loop
|
|
||||||
go w.Work()
|
|
||||||
wg.Add(1)
|
|
||||||
// calling Echo
|
|
||||||
w.Echo([]byte("Hello"))
|
|
||||||
// Waiting results
|
|
||||||
wg.Wait()
|
|
||||||
|
|
||||||
// tear down
|
|
||||||
w.Close()
|
|
||||||
}
|
|
@ -1,36 +1,18 @@
|
|||||||
package worker
|
package worker
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"bytes"
|
|
||||||
"flag"
|
|
||||||
"os"
|
|
||||||
"sync"
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
var (
|
var worker *Worker
|
||||||
worker *Worker
|
|
||||||
runIntegrationTests bool
|
|
||||||
)
|
|
||||||
|
|
||||||
func init() {
|
func init() {
|
||||||
worker = New(Unlimited)
|
worker = New(Unlimited)
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestMain(m *testing.M) {
|
|
||||||
integrationsTestFlag := flag.Bool("integration", false, "Run the integration tests (in addition to the unit tests)")
|
|
||||||
if integrationsTestFlag != nil {
|
|
||||||
runIntegrationTests = *integrationsTestFlag
|
|
||||||
}
|
|
||||||
code := m.Run()
|
|
||||||
os.Exit(code)
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestWorkerErrNoneAgents(t *testing.T) {
|
func TestWorkerErrNoneAgents(t *testing.T) {
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
err := worker.Ready()
|
err := worker.Ready()
|
||||||
if err != ErrNoneAgents {
|
if err != ErrNoneAgents {
|
||||||
t.Error("ErrNoneAgents expected.")
|
t.Error("ErrNoneAgents expected.")
|
||||||
@ -38,9 +20,6 @@ func TestWorkerErrNoneAgents(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestWorkerAddServer(t *testing.T) {
|
func TestWorkerAddServer(t *testing.T) {
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
t.Log("Add local server 127.0.0.1:4730.")
|
t.Log("Add local server 127.0.0.1:4730.")
|
||||||
if err := worker.AddServer(Network, "127.0.0.1:4730"); err != nil {
|
if err := worker.AddServer(Network, "127.0.0.1:4730"); err != nil {
|
||||||
t.Error(err)
|
t.Error(err)
|
||||||
@ -53,9 +32,6 @@ func TestWorkerAddServer(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestWorkerErrNoneFuncs(t *testing.T) {
|
func TestWorkerErrNoneFuncs(t *testing.T) {
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
err := worker.Ready()
|
err := worker.Ready()
|
||||||
if err != ErrNoneFuncs {
|
if err != ErrNoneFuncs {
|
||||||
t.Error("ErrNoneFuncs expected.")
|
t.Error("ErrNoneFuncs expected.")
|
||||||
@ -67,9 +43,6 @@ func foobar(job Job) ([]byte, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestWorkerAddFunction(t *testing.T) {
|
func TestWorkerAddFunction(t *testing.T) {
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
if err := worker.AddFunc("foobar", foobar, 0); err != nil {
|
if err := worker.AddFunc("foobar", foobar, 0); err != nil {
|
||||||
t.Error(err)
|
t.Error(err)
|
||||||
}
|
}
|
||||||
@ -83,18 +56,12 @@ func TestWorkerAddFunction(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestWorkerRemoveFunc(t *testing.T) {
|
func TestWorkerRemoveFunc(t *testing.T) {
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
if err := worker.RemoveFunc("foobar"); err != nil {
|
if err := worker.RemoveFunc("foobar"); err != nil {
|
||||||
t.Error(err)
|
t.Error(err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestWork(t *testing.T) {
|
func TestWork(t *testing.T) {
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
var wg sync.WaitGroup
|
var wg sync.WaitGroup
|
||||||
worker.JobHandler = func(job Job) error {
|
worker.JobHandler = func(job Job) error {
|
||||||
t.Logf("%s", job.Data())
|
t.Logf("%s", job.Data())
|
||||||
@ -111,76 +78,12 @@ func TestWork(t *testing.T) {
|
|||||||
wg.Wait()
|
wg.Wait()
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestLargeDataWork(t *testing.T) {
|
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
worker := New(Unlimited)
|
|
||||||
defer worker.Close()
|
|
||||||
|
|
||||||
if err := worker.AddServer(Network, "127.0.0.1:4730"); err != nil {
|
|
||||||
t.Error(err)
|
|
||||||
}
|
|
||||||
worker.Ready()
|
|
||||||
|
|
||||||
l := 5714
|
|
||||||
var wg sync.WaitGroup
|
|
||||||
|
|
||||||
bigdataHandler := func(job Job) error {
|
|
||||||
defer wg.Done()
|
|
||||||
if len(job.Data()) != l {
|
|
||||||
t.Errorf("expected length %d. got %d.", l, len(job.Data()))
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
if err := worker.AddFunc("bigdata", foobar, 0); err != nil {
|
|
||||||
defer wg.Done()
|
|
||||||
t.Error(err)
|
|
||||||
}
|
|
||||||
|
|
||||||
worker.JobHandler = bigdataHandler
|
|
||||||
|
|
||||||
worker.ErrorHandler = func(err error) {
|
|
||||||
t.Fatal("shouldn't have received an error")
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := worker.Ready(); err != nil {
|
|
||||||
t.Error(err)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
go worker.Work()
|
|
||||||
wg.Add(1)
|
|
||||||
|
|
||||||
// var cli *client.Client
|
|
||||||
// var err error
|
|
||||||
// if cli, err = client.New(client.Network, "127.0.0.1:4730"); err != nil {
|
|
||||||
// t.Fatal(err)
|
|
||||||
// }
|
|
||||||
// cli.ErrorHandler = func(e error) {
|
|
||||||
// t.Error(e)
|
|
||||||
// }
|
|
||||||
|
|
||||||
// _, err = cli.Do("bigdata", bytes.Repeat([]byte("a"), l), client.JobLow, func(res *client.Response) {
|
|
||||||
// })
|
|
||||||
// if err != nil {
|
|
||||||
// t.Error(err)
|
|
||||||
// }
|
|
||||||
|
|
||||||
worker.Echo(bytes.Repeat([]byte("a"), l))
|
|
||||||
wg.Wait()
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestWorkerClose(t *testing.T) {
|
func TestWorkerClose(t *testing.T) {
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
worker.Close()
|
worker.Close()
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestWorkWithoutReady(t *testing.T) {
|
func TestWorkWithoutReady(t * testing.T){
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
other_worker := New(Unlimited)
|
other_worker := New(Unlimited)
|
||||||
|
|
||||||
if err := other_worker.AddServer(Network, "127.0.0.1:4730"); err != nil {
|
if err := other_worker.AddServer(Network, "127.0.0.1:4730"); err != nil {
|
||||||
@ -189,15 +92,15 @@ func TestWorkWithoutReady(t *testing.T) {
|
|||||||
if err := other_worker.AddFunc("gearman-go-workertest", foobar, 0); err != nil {
|
if err := other_worker.AddFunc("gearman-go-workertest", foobar, 0); err != nil {
|
||||||
t.Error(err)
|
t.Error(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
timeout := make(chan bool, 1)
|
timeout := make(chan bool, 1)
|
||||||
done := make(chan bool, 1)
|
done := make( chan bool, 1)
|
||||||
|
|
||||||
other_worker.JobHandler = func(j Job) error {
|
other_worker.JobHandler = func( j Job ) error {
|
||||||
if !other_worker.ready {
|
if( ! other_worker.ready ){
|
||||||
t.Error("Worker not ready as expected")
|
t.Error("Worker not ready as expected");
|
||||||
}
|
}
|
||||||
done <- true
|
done <-true
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
go func() {
|
go func() {
|
||||||
@ -205,15 +108,15 @@ func TestWorkWithoutReady(t *testing.T) {
|
|||||||
timeout <- true
|
timeout <- true
|
||||||
}()
|
}()
|
||||||
|
|
||||||
go func() {
|
go func(){
|
||||||
other_worker.Work()
|
other_worker.Work();
|
||||||
}()
|
}()
|
||||||
|
|
||||||
// With the all-in-one Work() we don't know if the
|
// With the all-in-one Work() we don't know if the
|
||||||
// worker is ready at this stage so we may have to wait a sec:
|
// worker is ready at this stage so we may have to wait a sec:
|
||||||
go func() {
|
go func(){
|
||||||
tries := 5
|
tries := 5
|
||||||
for tries > 0 {
|
for( tries > 0 ){
|
||||||
if other_worker.ready {
|
if other_worker.ready {
|
||||||
other_worker.Echo([]byte("Hello"))
|
other_worker.Echo([]byte("Hello"))
|
||||||
break
|
break
|
||||||
@ -224,27 +127,24 @@ func TestWorkWithoutReady(t *testing.T) {
|
|||||||
tries--
|
tries--
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
// determine if we've finished or timed out:
|
// determine if we've finished or timed out:
|
||||||
select {
|
select{
|
||||||
case <-timeout:
|
case <- timeout:
|
||||||
t.Error("Test timed out waiting for the worker")
|
t.Error("Test timed out waiting for the worker")
|
||||||
case <-done:
|
case <- done:
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestWorkWithoutReadyWithPanic(t *testing.T) {
|
func TestWorkWithoutReadyWithPanic(t * testing.T){
|
||||||
if !runIntegrationTests {
|
|
||||||
t.Skip("To run this test, use: go test -integration")
|
|
||||||
}
|
|
||||||
other_worker := New(Unlimited)
|
other_worker := New(Unlimited)
|
||||||
|
|
||||||
timeout := make(chan bool, 1)
|
timeout := make(chan bool, 1)
|
||||||
done := make(chan bool, 1)
|
done := make( chan bool, 1)
|
||||||
|
|
||||||
// Going to work with no worker setup.
|
// Going to work with no worker setup.
|
||||||
// when Work (hopefully) calls Ready it will get an error which should cause it to panic()
|
// when Work (hopefully) calls Ready it will get an error which should cause it to panic()
|
||||||
go func() {
|
go func(){
|
||||||
defer func() {
|
defer func() {
|
||||||
if err := recover(); err != nil {
|
if err := recover(); err != nil {
|
||||||
done <- true
|
done <- true
|
||||||
@ -253,17 +153,17 @@ func TestWorkWithoutReadyWithPanic(t *testing.T) {
|
|||||||
t.Error("Work should raise a panic.")
|
t.Error("Work should raise a panic.")
|
||||||
done <- true
|
done <- true
|
||||||
}()
|
}()
|
||||||
other_worker.Work()
|
other_worker.Work();
|
||||||
}()
|
}()
|
||||||
go func() {
|
go func() {
|
||||||
time.Sleep(2 * time.Second)
|
time.Sleep(2 * time.Second)
|
||||||
timeout <- true
|
timeout <- true
|
||||||
}()
|
}()
|
||||||
|
|
||||||
select {
|
select{
|
||||||
case <-timeout:
|
case <- timeout:
|
||||||
t.Error("Test timed out waiting for the worker")
|
t.Error("Test timed out waiting for the worker")
|
||||||
case <-done:
|
case <- done:
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user