feature added to MockServer #5

Author: nathanaelle

.vscode/settings.json

@@ -2,7 +2,6 @@
 	"go.testOnSave": true,
 	"go.coverOnSave": true,
 	"go.coverageOptions": "showUncoveredCodeOnly",
-	"go.inferGopath": false,
 	"go.testTimeout": "5s",
 	"editor.tabSize": 8,
 	"editor.detectIndentation": false,

client-rr.go

@@ -27,7 +27,7 @@ func RoundRobinClient(ctx context.Context, debug *log.Logger) Client {
 	return c
 }
 
-func (c *rrServer) receivers() (<-chan Message, context.Context) {
+func (c *rrServer) Receivers() (<-chan Message, context.Context) {
 	return nil, nil
 }
 
@@ -64,13 +64,13 @@ func (c *rrServer) Submit(req Task) Task {
 	return cli.Submit(req)
 }
 
-func (c *rrServer) assignTask(tid TaskID) {
+func (c *rrServer) AssignTask(tid TaskID) {
 }
 
-func (c *rrServer) getTask(tid TaskID) Task {
+func (c *rrServer) GetTask(tid TaskID) Task {
 	return nil
 }
 
-func (c *rrServer) extractTask(tid TaskID) Task {
+func (c *rrServer) ExtractTask(tid TaskID) Task {
 	return nil
 }

client-single.go

@@ -14,6 +14,7 @@ type (
 		mQueue     chan Message
 		reqQueue   []Task
 		climutex   *sync.Mutex
+		reply      PacketEmiter
 	}
 )
 
@@ -30,7 +31,7 @@ func SingleServerClient(ctx context.Context, debug *log.Logger) Client {
 	return c
 }
 
-func (c *singleServer) receivers() (<-chan Message, context.Context) {
+func (c *singleServer) Receivers() (<-chan Message, context.Context) {
 	return c.mQueue, c.ctx
 }
 
@@ -50,9 +51,9 @@ func (c *singleServer) AddServers(servers ...Conn) {
 
 	c.configured = true
 
-	for _, server := range servers {
-		c.addServer(server)
-	}
+	reply, _ := c.addServer(servers[0])
+	c.reply = reply
+
 	return
 }
 
@@ -62,22 +63,20 @@ func (c *singleServer) Submit(req Task) Task {
 
 	c.reqQueue = append(c.reqQueue, req)
 
-	for _, s := range c.listServers() {
-		c.sendTo(s, req.Packet())
-	}
+	c.reply.Send(req.Packet())
 
 	return req
 }
 
-func (c *singleServer) assignTask(tid TaskID) {
+func (c *singleServer) AssignTask(tid TaskID) {
 	c.climutex.Lock()
 	defer c.climutex.Unlock()
 
 	c.jobs[tid.String()] = c.reqQueue[0]
 	c.reqQueue = c.reqQueue[1:]
 }
 
-func (c *singleServer) getTask(tid TaskID) Task {
+func (c *singleServer) GetTask(tid TaskID) Task {
 	c.climutex.Lock()
 	defer c.climutex.Unlock()
 
@@ -87,7 +86,7 @@ func (c *singleServer) getTask(tid TaskID) Task {
 	return NilTask
 }
 
-func (c *singleServer) extractTask(tid TaskID) Task {
+func (c *singleServer) ExtractTask(tid TaskID) Task {
 	c.climutex.Lock()
 	defer c.climutex.Unlock()
 

client.go

@@ -14,18 +14,18 @@ type (
 		Submit(Task) Task
 		Close() error
 
-		assignTask(tid TaskID)
-		getTask(TaskID) Task
-		extractTask(TaskID) Task
-		receivers() (<-chan Message, context.Context)
+		AssignTask(TaskID)
+		GetTask(TaskID) Task
+		ExtractTask(TaskID) Task
+		Receivers() (<-chan Message, context.Context)
 	}
 )
 
 func clientLoop(c Client, dbg *log.Logger) {
 	var tid TaskID
 	var err error
 
-	mQueue, ctx := c.receivers()
+	mQueue, ctx := c.Receivers()
 
 	for {
 		select {
@@ -53,23 +53,23 @@ func clientLoop(c Client, dbg *log.Logger) {
 					debug(dbg, "CLI\tprotocol.JobCreated TID [%s] err : %v\n", string(msg.Pkt.At(0).Bytes()), err)
 					panic(err)
 				}
-				c.assignTask(tid)
+				c.AssignTask(tid)
 
 			case protocol.WorkData, protocol.WorkWarning, protocol.WorkStatus:
 				if err = tid.Cast(msg.Pkt.At(0)); err != nil {
 					debug(dbg, "CLI\t%s TID [%s] err : %v\n", msg.Pkt.Cmd(), string(msg.Pkt.At(0).Bytes()), err)
 					panic(err)
 				}
 
-				c.getTask(tid).Handle(msg.Pkt)
+				c.GetTask(tid).Handle(msg.Pkt)
 
 			case protocol.WorkComplete, protocol.WorkFail, protocol.WorkException:
 				if err = tid.Cast(msg.Pkt.At(0)); err != nil {
 					debug(dbg, "CLI\t%s TID [%s] err : %v\n", msg.Pkt.Cmd(), string(msg.Pkt.At(0).Bytes()), err)
 					panic(err)
 				}
 
-				c.extractTask(tid).Handle(msg.Pkt)
+				c.ExtractTask(tid).Handle(msg.Pkt)
 
 			case protocol.StatusRes:
 				panic("status_res not wrote")

error.go

@@ -26,7 +26,7 @@ var (
 )
 
 func (e *ExceptionError) Error() string {
-	return fmt.Sprintf("Error [%x]", e.Payload)
+	return fmt.Sprintf("Error [%q]", e.Payload)
 }
 
 func (e *IncoherentError) Error() string {

message.go

@@ -0,0 +1,55 @@
+package gearman // import "github.com/nathanaelle/gearman"
+
+import (
+	"sync"
+
+	"github.com/nathanaelle/gearman/v2/protocol"
+)
+
+type (
+	// PacketEmiter is a interface to send packet to a socket pool
+	PacketEmiter interface {
+		Send(...protocol.Packet)
+	}
+
+	// Message is the structure for communication between a pool and a Client or a Worker
+	Message struct {
+		Reply  PacketEmiter
+		Server Conn
+		Pkt    protocol.Packet
+	}
+
+	chanPacketEmiter struct {
+		c chan<- protocol.Packet
+	}
+
+	funcPacketEmiter struct {
+		lock         *sync.Mutex
+		packetEmiter func(protocol.Packet, PacketEmiter)
+	}
+)
+
+var _ PacketEmiter = chanPacketEmiter{}
+var _ PacketEmiter = &funcPacketEmiter{}
+
+func newChanPacketEmiter(c chan<- protocol.Packet) PacketEmiter {
+	return chanPacketEmiter{c}
+}
+
+func (cpe chanPacketEmiter) Send(pkts ...protocol.Packet) {
+	for _, pkt := range pkts {
+		cpe.c <- pkt
+	}
+}
+
+func newFuncPacketEmiter(pe func(protocol.Packet, PacketEmiter)) PacketEmiter {
+	return &funcPacketEmiter{
+		packetEmiter: pe,
+	}
+}
+
+func (fpe *funcPacketEmiter) Send(pkts ...protocol.Packet) {
+	for _, pkt := range pkts {
+		fpe.packetEmiter(pkt, fpe)
+	}
+}

mock.go

@@ -1,4 +1,4 @@
-package gearman
+package gearman // import "github.com/nathanaelle/gearman"
 
 import (
 	"bytes"
@@ -28,10 +28,12 @@ func NewMockServer() *MockServer {
 	}
 }
 
+// AddServers implements Client.AddServers and Worker.AddServers
 func (mc *MockServer) AddServers(...Conn) {
 
 }
 
+// AddHandler implements Worker.AddHandler
 func (mc *MockServer) AddHandler(name string, job Job) Worker {
 	mc.lock.Lock()
 	defer mc.lock.Unlock()
@@ -41,6 +43,7 @@ func (mc *MockServer) AddHandler(name string, job Job) Worker {
 	return mc
 }
 
+// DelHandler implements Worker.DelHandler
 func (mc *MockServer) DelHandler(name string) Worker {
 	mc.lock.Lock()
 	defer mc.lock.Unlock()
@@ -50,6 +53,7 @@ func (mc *MockServer) DelHandler(name string) Worker {
 	return mc
 }
 
+// DelAllHandlers implements Worker.DelAllHandlers
 func (mc *MockServer) DelAllHandlers() Worker {
 	mc.lock.Lock()
 	defer mc.lock.Unlock()
@@ -59,6 +63,7 @@ func (mc *MockServer) DelAllHandlers() Worker {
 	return mc
 }
 
+// GetHandler implements Worker.GetHandler
 func (mc *MockServer) GetHandler(name string) Job {
 	mc.lock.Lock()
 	defer mc.lock.Unlock()
@@ -70,57 +75,76 @@ func (mc *MockServer) GetHandler(name string) Job {
 	return FailJob
 }
 
+// Receivers implements Client.Receivers and Worker.Receivers
 func (mc *MockServer) Receivers() (<-chan Message, context.Context) {
 	return nil, nil
 }
 
+// Close implements Client.Close and Worker.Close
 func (mc *MockServer) Close() error {
 	return nil
 }
 
+// Submit implements Client.Submit
 func (mc *MockServer) Submit(req Task) Task {
 	pkt := req.Packet()
 
 	switch pkt.Cmd() {
 	case protocol.SubmitJob:
-		reply := make(chan protocol.Packet, 5)
+		reply := mockPacketEmiter(req)
 
 		go runWorker(mc.GetHandler(string(pkt.At(0).Bytes())), bytes.NewReader(pkt.At(2).Bytes()), reply, TaskID{})
 
-		go func() {
-			for res := range reply {
-				switch res.Cmd() {
-				case protocol.WorkCompleteWorker:
-					taskRes, _ := protocol.WorkComplete.Borrow(res)
-					go req.Handle(taskRes)
-					close(reply)
-					break
-
-				default:
-					log.Fatalf("res unknown: %v %q", res.Cmd(), res.Payload())
-				}
-			}
-		}()
-
 	default:
 		log.Fatalf("unknown: %v %q", pkt.Cmd(), pkt.Payload())
 	}
 
 	return req
 }
 
-func (mc *MockServer) assignTask(tid TaskID) {
+// AssignTask implements Client.AssignTask
+func (mc *MockServer) AssignTask(tid TaskID) {
 
 }
 
-func (mc *MockServer) getTask(TaskID) Task {
+// GetTask implements Client.GetTask
+func (mc *MockServer) GetTask(TaskID) Task {
 	return nil
 }
 
-func (mc *MockServer) extractTask(TaskID) Task {
+// ExtractTask implements Client.ExtractTask
+func (mc *MockServer) ExtractTask(TaskID) Task {
 	return nil
 }
 
-func (mc *MockServer) receivers() (<-chan Message, context.Context) {
-	return nil, nil
+func mockPacketEmiter(req Task) PacketEmiter {
+	lock := sync.Mutex{}
+
+	return newFuncPacketEmiter(func(res protocol.Packet, fpe PacketEmiter) {
+		lock.Lock()
+		defer lock.Unlock()
+
+		switch res.Cmd() {
+		case protocol.WorkCompleteWorker:
+			taskRes, _ := protocol.WorkComplete.Borrow(res)
+			go req.Handle(taskRes)
+			return
+
+		case protocol.WorkFailWorker:
+			taskRes, _ := protocol.WorkFail.Borrow(res)
+			go req.Handle(taskRes)
+			return
+
+		case protocol.WorkExceptionWorker:
+			taskRes, _ := protocol.WorkException.Borrow(res)
+			go req.Handle(taskRes)
+			return
+
+		case protocol.WorkDataWorker:
+			taskRes, _ := protocol.WorkData.Borrow(res)
+			go req.Handle(taskRes)
+			return
+		}
+		log.Fatalf("res unknown: %v %q", res.Cmd(), res.Payload())
+	})
 }

mock_test.go

@@ -2,6 +2,7 @@ package gearman
 
 import (
 	"context"
+	"errors"
 	"io"
 	"sync"
 	"testing"
@@ -31,6 +32,13 @@ func mockServerClient(cancel context.CancelFunc, wg *sync.WaitGroup, cli Client,
 	if !validResult(t, []byte("tset"), nil)(r.Value()) {
 		return
 	}
+
+	r = cli.Submit(NewTask("esrever", []byte("test")))
+
+	if !validResult(t, nil, errors.New(`Error ["job doesn't exist"]`))(r.Value()) {
+		return
+	}
+
 }
 
 func mockServerWorker(ctx context.Context, wg *sync.WaitGroup, wkr Worker, t *testing.T) {