.travis.yml

@@ -1,6 +1,6 @@
 language: go
 go:
-  - 1.4
+  - 1.2
 
 before_install:
   - sudo apt-get remove -y gearman-job-server

README.md

@@ -37,9 +37,7 @@ Usage
 
 ## Worker
 
-```go
+	// Limit number of concurrent jobs execution. Use worker.Unlimited (0) if you want no limitation.
-// Limit number of concurrent jobs execution. 
-// Use worker.Unlimited (0) if you want no limitation.
     w := worker.New(worker.OneByOne)
     w.ErrHandler = func(e error) {
         log.Println(e)
@@ -49,17 +47,15 @@ w.AddServer("127.0.0.1:4730")
     w.AddFunc("ToUpper", ToUpper, worker.Unlimited)
 	// This will give a timeout of 5 seconds
     w.AddFunc("ToUpperTimeOut5", ToUpper, 5)
-
 	if err := w.Ready(); err != nil {
 		log.Fatal(err)
 		return
 	}
 	go w.Work()
-```
+	
 
 ## Client
 
-```go
 	// ...
 	c, err := client.New("tcp4", "127.0.0.1:4730")
     // ... error handling
@@ -71,12 +67,11 @@ echo := []byte("Hello\x00 world")
 	echomsg, err := c.Echo(echo)
 	// ... error handling
     log.Println(string(echomsg))
-jobHandler := func(resp *client.Response) {
+    jobHandler := func(job *client.Job) {
-	log.Printf("%s", resp.Data)
+        log.Printf("%s", job.Data)
     }
-handle, err := c.Do("ToUpper", echo, client.JobNormal, jobHandler)
+    handle, err := c.Do("ToUpper", echo, client.JOB_NORMAL, jobHandler)
 	// ...	
-```
 
 Branches
 ========
@@ -92,15 +87,10 @@ __Use at your own risk!__
 Contributors
 ============
 
-Great thanks to all of you for your support and interest!
-
 (_Alphabetic order_)
  
  * [Alex Zylman](https://github.com/azylman)
- * [C.R. Kirkwood-Watts](https://github.com/kirkwood)
  * [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)
  * [jake](https://github.com/jbaikge)
  * [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)
  * [miraclesu](https://github.com/miraclesu)
  * [Paul Mach](https://github.com/paulmach)
- * [Randall McPherson](https://github.com/rlmcpherson)
  * [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) &lt;<mikespook@gmail.com>&gt; [@Twitter](http://twitter.com/mikespook)
 
 Open Source - MIT Software License
 ==================================

client/client.go

@@ -6,11 +6,6 @@ import (
 	"bufio"
 	"net"
 	"sync"
-	"time"
-)
-
-var (
-	DefaultTimeout time.Duration = time.Second
 )
 
 // One client connect to one server.
@@ -18,63 +13,24 @@ var (
 type Client struct {
 	sync.Mutex
 
-	net, addr    string
+	net, addr, lastcall string
-	innerHandler *responseHandlerMap
+	respHandler         map[string]ResponseHandler
+	innerHandler        map[string]ResponseHandler
 	in                  chan *Response
 	conn                net.Conn
 	rw                  *bufio.ReadWriter
 
-	ResponseTimeout time.Duration // response timeout for do()
-
 	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) {
 	client = &Client{
 		net:          network,
 		addr:         addr,
-		innerHandler:    newResponseHandlerMap(),
+		respHandler:  make(map[string]ResponseHandler, queueSize),
+		innerHandler: make(map[string]ResponseHandler, queueSize),
 		in:           make(chan *Response, queueSize),
-		ResponseTimeout: DefaultTimeout,
 	}
 	client.conn, err = net.Dial(client.net, client.addr)
 	if err != nil {
@@ -148,7 +104,6 @@ ReadLoop:
 		}
 		if len(leftdata) > 0 { // some data left for processing
 			data = append(leftdata, data...)
-			leftdata = nil
 		}
 		for {
 			l := len(data)
@@ -172,25 +127,27 @@ ReadLoop:
 }
 
 func (client *Client) processLoop() {
-	rhandlers := map[string]ResponseHandler{}
 	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:
-			client.handleInner("s"+resp.Handle, resp, nil)
-		case dtJobCreated:
-			client.handleInner("c", resp, rhandlers)
-		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:
-			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,
-	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 {
 		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 {
 			err = getError(resp.Data)
-			result <- handleOrError{"", err}
 			return
 		}
 		handle = resp.Handle
-		result <- handleOrError{handle, nil}
+	}
-	}, h)
+	id := IdGen.Id()
 	req := getJob(id, []byte(funcname), data)
 	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
 }
 
@@ -257,7 +204,19 @@ func (client *Client) do(funcname string, data []byte,
 // flag can be set to: JobLow, JobNormal and JobHigh
 func (client *Client) Do(funcname string, data []byte,
 	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
 }
 
@@ -265,25 +224,38 @@ func (client *Client) Do(funcname string, data []byte,
 // flag can be set to: JobLow, JobNormal and JobHigh
 func (client *Client) DoBg(funcname string, data []byte,
 	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
 }
 
-// Status gets job status from job server.
+// Get job status from job server.
 func (client *Client) Status(handle string) (status *Status, err error) {
 	if client.conn == nil {
 		return nil, ErrLostConn
 	}
 	var mutex sync.Mutex
 	mutex.Lock()
-	client.innerHandler.put("s"+handle, func(resp *Response) {
+	client.lastcall = "s" + handle
+	client.innerHandler["s"+handle] = func(resp *Response) {
 		defer mutex.Unlock()
 		var err error
 		status, err = resp._status()
 		if err != nil {
 			client.err(err)
 		}
-	})
+	}
 	req := getRequest()
 	req.DataType = dtGetStatus
 	req.Data = []byte(handle)
@@ -299,13 +271,14 @@ func (client *Client) Echo(data []byte) (echo []byte, err error) {
 	}
 	var mutex sync.Mutex
 	mutex.Lock()
-	client.innerHandler.put("e", func(resp *Response) {
+	client.innerHandler["e"] = func(resp *Response) {
 		echo = resp.Data
 		mutex.Unlock()
-	})
+	}
 	req := getRequest()
 	req.DataType = dtEchoReq
 	req.Data = data
+	client.lastcall = "e"
 	client.write(req)
 	mutex.Lock()
 	return
@@ -321,40 +294,3 @@ func (client *Client) Close() (err error) {
 	}
 	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
-}

client/client_test.go

@@ -1,39 +1,16 @@
 package client
 
 import (
-	"crypto/md5"
-	"encoding/hex"
-	"errors"
-	"flag"
-	"fmt"
-	"os"
 	"testing"
-	"time"
 )
 
 const (
 	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) {
-	if !runIntegrationTests {
-		t.Skip("To run this test, use: go test -integration")
-	}
 	t.Log("Add local server 127.0.0.1:4730")
 	var err error
 	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) {
-	if !runIntegrationTests {
-		t.Skip("To run this test, use: go test -integration")
-	}
 	echo, err := client.Echo([]byte(TestStr))
 	if err != nil {
 		t.Error(err)
@@ -60,9 +34,6 @@ func TestClientEcho(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)
 	if err != nil {
 		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) {
-	if !runIntegrationTests {
-		t.Skip("To run this test, use: go test -integration")
-	}
 	jobHandler := func(job *Response) {
 		str := string(job.Data)
 		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) {
-	if !runIntegrationTests {
-		t.Skip("To run this test, use: go test -integration")
-	}
 	status, err := client.Status("handle not exists")
 	if err != nil {
 		t.Error(err)
@@ -368,9 +105,6 @@ func TestClientStatus(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 {
 		t.Error(err)
 	}

client/common.go

@@ -5,7 +5,7 @@ const (
 	// queue size
 	queueSize = 8
 	// read buffer size
-	bufferSize = 8192
+	bufferSize = 1024
 	// min packet length
 	minPacketLength = 12
 
@@ -51,7 +51,6 @@ const (
 	dtSubmitJobLowBg  = 34
 
 	WorkComplate  = dtWorkComplete
-	WorkComplete  = dtWorkComplete
 	WorkData      = dtWorkData
 	WorkStatus    = dtWorkStatus
 	WorkWarning   = dtWorkWarning

client/error.go

@@ -9,7 +9,6 @@ import (
 var (
 	ErrWorkWarning   = errors.New("Work warning")
 	ErrInvalidData   = errors.New("Invalid data")
-	ErrInvalidId     = errors.New("Invalid ID")
 	ErrWorkFail      = errors.New("Work fail")
 	ErrWorkException = errors.New("Work exeption")
 	ErrDataType      = errors.New("Invalid data type")

client/id.go

@@ -32,7 +32,7 @@ func (ai *autoincId) Id() string {
 	return strconv.FormatInt(next, 10)
 }
 
-// NewAutoIncId returns an autoincrement ID generator
+// Return an autoincrement ID generator
 func NewAutoIncId() IdGenerator {
 	// we'll consider the nano fraction of a second at startup unique
 	// and count up from there.

client/pool.go

@@ -12,17 +12,19 @@ const (
 
 var (
 	ErrNotFound    = errors.New("Server Not Found")
+	SelectWithRate = selectWithRate
+	SelectRandom   = selectRandom
 )
 
-type PoolClient struct {
+type poolClient struct {
 	*Client
 	Rate  int
 	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) {
 	total := 0
 	for _, item := range pool {
@@ -34,7 +36,7 @@ func SelectWithRate(pool map[string]*PoolClient,
 	return last
 }
 
-func SelectRandom(pool map[string]*PoolClient,
+func selectRandom(pool map[string]*poolClient,
 	last string) (addr string) {
 	r := rand.Intn(len(pool))
 	i := 0
@@ -50,17 +52,17 @@ func SelectRandom(pool map[string]*PoolClient,
 type Pool struct {
 	SelectionHandler SelectionHandler
 	ErrorHandler     ErrorHandler
-	Clients          map[string]*PoolClient
 
+	clients map[string]*poolClient
 	last    string
 
 	mutex sync.Mutex
 }
 
-// NewPool returns a new pool.
+// Return a new pool.
 func NewPool() (pool *Pool) {
 	return &Pool{
-		Clients:          make(map[string]*PoolClient, poolSize),
+		clients:          make(map[string]*poolClient, poolSize),
 		SelectionHandler: SelectWithRate,
 	}
 }
@@ -69,16 +71,16 @@ func NewPool() (pool *Pool) {
 func (pool *Pool) Add(net, addr string, rate int) (err error) {
 	pool.mutex.Lock()
 	defer pool.mutex.Unlock()
-	var item *PoolClient
+	var item *poolClient
 	var ok bool
-	if item, ok = pool.Clients[addr]; ok {
+	if item, ok = pool.clients[addr]; ok {
 		item.Rate = rate
 	} else {
 		var client *Client
 		client, err = New(net, addr)
 		if err == nil {
-			item = &PoolClient{Client: client, Rate: rate}
+			item = &poolClient{Client: client, Rate: rate}
-			pool.Clients[addr] = item
+			pool.clients[addr] = item
 		}
 	}
 	return
@@ -88,7 +90,7 @@ func (pool *Pool) Add(net, addr string, rate int) (err error) {
 func (pool *Pool) Remove(addr string) {
 	pool.mutex.Lock()
 	defer pool.mutex.Unlock()
-	delete(pool.Clients, addr)
+	delete(pool.clients, addr)
 }
 
 func (pool *Pool) Do(funcname string, data []byte,
@@ -111,10 +113,10 @@ func (pool *Pool) DoBg(funcname string, data []byte,
 	return
 }
 
-// Status gets job status from job server.
+// Get job status from job server.
 // !!!Not fully tested.!!!
 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()
 		defer client.Unlock()
 		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.
 func (pool *Pool) Echo(addr string, data []byte) (echo []byte, err error) {
-	var client *PoolClient
+	var client *poolClient
 	if addr == "" {
 		client = pool.selectServer()
 	} else {
 		var ok bool
-		if client, ok = pool.Clients[addr]; !ok {
+		if client, ok = pool.clients[addr]; !ok {
 			err = ErrNotFound
 			return
 		}
@@ -145,18 +147,18 @@ func (pool *Pool) Echo(addr string, data []byte) (echo []byte, err error) {
 // Close
 func (pool *Pool) Close() (err map[string]error) {
 	err = make(map[string]error)
-	for _, c := range pool.Clients {
+	for _, c := range pool.clients {
 		err[c.addr] = c.Close()
 	}
 	return
 }
 
 // selecting server
-func (pool *Pool) selectServer() (client *PoolClient) {
+func (pool *Pool) selectServer() (client *poolClient) {
 	for client == nil {
-		addr := pool.SelectionHandler(pool.Clients, pool.last)
+		addr := pool.SelectionHandler(pool.clients, pool.last)
 		var ok bool
-		if client, ok = pool.Clients[addr]; ok {
+		if client, ok = pool.clients[addr]; ok {
 			pool.last = addr
 			break
 		}

client/pool_test.go

@@ -9,9 +9,6 @@ var (
 )
 
 func TestPoolAdd(t *testing.T) {
-	if !runIntegrationTests {
-		t.Skip("To run this test, use: go test -integration")
-	}
 	t.Log("Add servers")
 	c := 2
 	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)
 		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) {
-	if !runIntegrationTests {
-		t.Skip("To run this test, use: go test -integration")
-	}
 	echo, err := pool.Echo("", []byte(TestStr))
 	if err != nil {
 		t.Error(err)
@@ -47,9 +41,6 @@ func TestPoolEcho(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",
 		[]byte("abcdef"), JobLow)
 	if err != nil {
@@ -64,9 +55,6 @@ func TestPoolDoBg(t *testing.T) {
 }
 
 func TestPoolDo(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" {
@@ -89,9 +77,6 @@ func TestPoolDo(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")
 	if err != nil {
 		t.Error(err)
@@ -129,9 +114,6 @@ func TestPoolStatus(t *testing.T) {
 }
 
 func TestPoolClose(t *testing.T) {
-	if !runIntegrationTests {
-		t.Skip("To run this test, use: go test -integration")
-	}
 	return
 	if err := pool.Close(); err != nil {
 		t.Error(err)

client/response.go

@@ -76,7 +76,7 @@ func decodeResponse(data []byte) (resp *Response, l int, err error) {
 	case dtJobCreated:
 		resp.Handle = string(dt)
 	case dtStatusRes, dtWorkData, dtWorkWarning, dtWorkStatus,
-		dtWorkComplete, dtWorkException:
+		dtWorkComplete, dtWorkFail, dtWorkException:
 		s := bytes.SplitN(dt, []byte{'\x00'}, 2)
 		if len(s) >= 2 {
 			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)
 			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:
 		fallthrough
 	default:

example/client/client.go

@@ -3,8 +3,8 @@ package main
 import (
 	"github.com/mikespook/gearman-go/client"
 	"log"
-	"os"
 	"sync"
+	"os"
 )
 
 func main() {

example/pl/worker_multi.pl

@@ -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;

example/worker/worker.go

@@ -1,14 +1,13 @@
 package main
 
 import (
+	"github.com/mikespook/gearman-go/worker"
+	"github.com/mikespook/golib/signal"
 	"log"
-	"net"
 	"os"
 	"strings"
 	"time"
-
+	"net"
-	"github.com/mikespook/gearman-go/worker"
-	"github.com/mikespook/golib/signal"
 )
 
 func ToUpper(job worker.Job) ([]byte, error) {
@@ -69,6 +68,7 @@ func main() {
 		return
 	}
 	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()
 }

gearman.go

@@ -17,3 +17,8 @@ in an easy way.
 	import "github.com/mikespook/gearman-go/worker"
 */
 package gearman
+
+import (
+	_ "github.com/mikespook/gearman-go/client"
+	_ "github.com/mikespook/gearman-go/worker"
+)

worker/agent.go

@@ -2,9 +2,6 @@ package worker
 
 import (
 	"bufio"
-	"bytes"
-	"encoding/binary"
-	"io"
 	"net"
 	"sync"
 )
@@ -49,26 +46,20 @@ func (a *agent) work() {
 			a.worker.err(err.(error))
 		}
 	}()
-
 	var inpack *inPack
 	var l int
 	var err error
 	var data, leftdata []byte
 	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.Temporary() {
 					continue
 				}else{
-					a.disconnect_error(err)
+					a.worker.err(err)
-					// else - we're probably dc'ing due to a Close()
-
 					break
 				}
 				
-			} else if err == io.EOF {
-				a.disconnect_error(err)
-				break
 			}
 			a.worker.err(err)
 			// If it is unexpected error and the connection wasn't
@@ -90,43 +81,19 @@ func (a *agent) work() {
 			leftdata = data
 			continue
 		}
-		for {
 		if inpack, l, err = decodeInPack(data); err != nil {
 			a.worker.err(err)
 			leftdata = data
-				break
+			continue
-			} 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 = data[l:]
 		}
 	}
 }
-	}
-}
-
-func (a *agent) disconnect_error(err error) {
-	a.Lock()
-	defer a.Unlock()
-
-	if a.conn != nil {
-		err = &WorkerDisconnectError{
-			err:   err,
-			agent: a,
-		}
-		a.worker.err(err)
-	}
-}
 
 func (a *agent) Close() {
 	a.Lock()
@@ -140,17 +107,9 @@ func (a *agent) Close() {
 func (a *agent) Grab() {
 	a.Lock()
 	defer a.Unlock()
-	a.grab()
-}
-
-func (a *agent) grab() bool {
-	if a.worker.closed != nil {
-		return false
-	}
 	outpack := getOutPack()
 	outpack.dataType = dtGrabJobUniq
 	a.write(outpack)
-	return true
 }
 
 func (a *agent) PreSleep() {
@@ -161,51 +120,21 @@ func (a *agent) PreSleep() {
 	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
-func (a *agent) read() (data []byte, err error) {
+func (a *agent) read(length int) (data []byte, err error) {
 	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
-	if n, err = a.rw.Read(tmp); err != nil {
 			return
 		}
-	dl := int(binary.BigEndian.Uint32(tmp[8:12]))
+		data = append(data, buf[0:n]...)
-
+		if n < bufferSize {
-	// write what we read so far
+			break
-	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.
@@ -220,10 +149,3 @@ func (a *agent) write(outpack *outPack) (err error) {
 	}
 	return a.rw.Flush()
 }
-
-// Write with lock
-func (a *agent) Write(outpack *outPack) (err error) {
-	a.Lock()
-	defer a.Unlock()
-	return a.write(outpack)
-}

worker/example_test.go

@@ -53,4 +53,5 @@ func ExampleWorker() {
 	w.Echo([]byte("Hello"))
 	// Waiting results
 	wg.Wait()
+	// Output: Hello
 }

worker/worker.go

@@ -3,8 +3,8 @@
 package worker
 
 import (
+	"encoding/binary"
 	"fmt"
-	"strconv"
 	"sync"
 	"time"
 )
@@ -23,17 +23,14 @@ type Worker struct {
 	in      chan *inPack
 	running bool
 	ready bool
-	jobLeftNum int64
 
 	Id           string
 	ErrorHandler ErrorHandler
 	JobHandler   JobHandler
 	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
 // 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'.
 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.
 func (worker *Worker) broadcast(outpack *outPack) {
 	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.
 func (worker *Worker) AddFunc(funcname string,
 	f JobFunc, timeout uint32) (err error) {
@@ -97,11 +94,6 @@ func (worker *Worker) AddFunc(funcname string,
 
 // inner add
 func (worker *Worker) addFunc(funcname string, timeout uint32) {
-	outpack := prepFuncOutpack(funcname, timeout)
-	worker.broadcast(outpack)
-}
-
-func prepFuncOutpack(funcname string, timeout uint32) *outPack {
 	outpack := getOutPack()
 	if timeout == 0 {
 		outpack.dataType = dtCanDo
@@ -109,17 +101,15 @@ func prepFuncOutpack(funcname string, timeout uint32) *outPack {
 	} else {
 		outpack.dataType = dtCanDoTimeout
 		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))
 		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) {
 	worker.Lock()
 	defer worker.Unlock()
@@ -150,11 +140,6 @@ func (worker *Worker) handleInPack(inpack *inPack) {
 		inpack.a.Grab()
 	case dtJobAssign, dtJobAssignUniq:
 		go func() {
-			go func() {
-				worker.incrExecJobNum()
-				defer func() {
-					worker.decrExecJobNum()
-				}()
 			if err := worker.exec(inpack); err != nil {
 				worker.err(err)
 			}
@@ -163,7 +148,6 @@ func (worker *Worker) handleInPack(inpack *inPack) {
 			worker.limit <- true
 		}
 		inpack.a.Grab()
-		}()
 	case dtError:
 		worker.err(inpack.Err())
 		fallthrough
@@ -195,7 +179,7 @@ func (worker *Worker) Ready() (err error) {
 	return
 }
 
-// Work start main loop (blocking)
+// Main loop, block here
 // Most of time, this should be evaluated in goroutine.
 func (worker *Worker) Work() {
 	if ! worker.ready {
@@ -206,29 +190,19 @@ func (worker *Worker) Work() {
 		}
 	}
 
-	worker.Lock()
+	defer func() {
+		for _, a := range worker.agents {
+			a.Close()
+		}
+	}()
 	worker.running = true
-	worker.Unlock()
-
 	for _, a := range worker.agents {
 		a.Grab()
 	}
-	// 执行任务(阻塞)
 	var inpack *inPack
 	for inpack = range worker.in {
 		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
@@ -243,22 +217,10 @@ func (worker *Worker) customeHandler(inpack *inPack) {
 // Close connection and exit main loop
 func (worker *Worker) Close() {
 	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
 		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)
 }
 
-// Reset removes all of functions.
+// Remove all of functions.
 // Both from the worker and job servers.
 func (worker *Worker) Reset() {
 	outpack := getOutPack()
@@ -288,23 +250,6 @@ func (worker *Worker) SetId(id string) {
 	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
 func (worker *Worker) exec(inpack *inPack) (err error) {
 	defer func() {
@@ -330,7 +275,7 @@ func (worker *Worker) exec(inpack *inPack) (err error) {
 	} else {
 		r = execTimeout(f.f, inpack, time.Duration(f.timeout)*time.Second)
 	}
-	//if worker.running {
+	if worker.running {
 		outpack := getOutPack()
 		if r.err == nil {
 			outpack.dataType = dtWorkComplete
@@ -344,22 +289,10 @@ func (worker *Worker) exec(inpack *inPack) (err error) {
 		}
 		outpack.handle = inpack.handle
 		outpack.data = r.data
-	_ = inpack.a.Write(outpack)
+		inpack.a.write(outpack)
-	if worker.leftJobs != nil {
-		worker.leftJobs <- struct{}{}
 	}
-	//}
 	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
 type result struct {
@@ -383,23 +316,3 @@ func execTimeout(f JobFunc, job Job, timeout time.Duration) (r *result) {
 	}
 	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
-}

worker/worker_disconnect_test.go

@@ -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() )
+	}
+}

worker/worker_racy_test.go

@@ -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()
-}

worker/worker_test.go

@@ -1,36 +1,18 @@
 package worker
 
 import (
-	"bytes"
-	"flag"
-	"os"
 	"sync"
 	"testing"
 	"time"
 )
 
-var (
+var worker *Worker
-	worker              *Worker
-	runIntegrationTests bool
-)
 
 func init() {
 	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) {
-	if !runIntegrationTests {
-		t.Skip("To run this test, use: go test -integration")
-	}
 	err := worker.Ready()
 	if err != ErrNoneAgents {
 		t.Error("ErrNoneAgents expected.")
@@ -38,9 +20,6 @@ func TestWorkerErrNoneAgents(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.")
 	if err := worker.AddServer(Network, "127.0.0.1:4730"); err != nil {
 		t.Error(err)
@@ -53,9 +32,6 @@ func TestWorkerAddServer(t *testing.T) {
 }
 
 func TestWorkerErrNoneFuncs(t *testing.T) {
-	if !runIntegrationTests {
-		t.Skip("To run this test, use: go test -integration")
-	}
 	err := worker.Ready()
 	if err != ErrNoneFuncs {
 		t.Error("ErrNoneFuncs expected.")
@@ -67,9 +43,6 @@ func foobar(job Job) ([]byte, error) {
 }
 
 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 {
 		t.Error(err)
 	}
@@ -83,18 +56,12 @@ func TestWorkerAddFunction(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 {
 		t.Error(err)
 	}
 }
 
 func TestWork(t *testing.T) {
-	if !runIntegrationTests {
-		t.Skip("To run this test, use: go test -integration")
-	}
 	var wg sync.WaitGroup
 	worker.JobHandler = func(job Job) error {
 		t.Logf("%s", job.Data())
@@ -111,76 +78,12 @@ func TestWork(t *testing.T) {
 	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) {
-	if !runIntegrationTests {
-		t.Skip("To run this test, use: go test -integration")
-	}
 	worker.Close()
 }
 
 func TestWorkWithoutReady(t * testing.T){
-	if !runIntegrationTests {
-		t.Skip("To run this test, use: go test -integration")
-	}
 	other_worker := New(Unlimited)
 
 	if err := other_worker.AddServer(Network, "127.0.0.1:4730"); err != nil {
@@ -194,8 +97,8 @@ func TestWorkWithoutReady(t *testing.T) {
 	done := make( chan bool, 1)
 
 	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
 		return nil
@@ -206,14 +109,14 @@ func TestWorkWithoutReady(t *testing.T) {
 	}()
 
 	go func(){
-		other_worker.Work()
+		other_worker.Work();
 	}()
 
 	// 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:
 	go func(){
 		tries := 5
-		for tries > 0 {
+		for( tries > 0 ){
 			if other_worker.ready {
 				other_worker.Echo([]byte("Hello"))
 				break
@@ -234,9 +137,6 @@ func TestWorkWithoutReady(t *testing.T) {
 }
 
 func TestWorkWithoutReadyWithPanic(t * testing.T){
-	if !runIntegrationTests {
-		t.Skip("To run this test, use: go test -integration")
-	}
 	other_worker := New(Unlimited)
 	
 	timeout := make(chan bool, 1)
@@ -253,7 +153,7 @@ func TestWorkWithoutReadyWithPanic(t *testing.T) {
 			t.Error("Work should raise a panic.")
 			done <- true
 		}()
-		other_worker.Work()
+		other_worker.Work();
 	}()
 	go func() {
 		time.Sleep(2 * time.Second)