refactoring worker side data package

This commit is contained in:
Xing Xing 2013-12-20 15:24:22 +08:00
parent e5179e3b5b
commit e9c29799fb
6 changed files with 27 additions and 34 deletions

View File

@ -9,7 +9,7 @@ import (
"encoding/binary"
)
// request
// Request from client
type request struct {
DataType uint32
Data []byte

View File

@ -3,7 +3,6 @@ package main
import (
"log"
"sync"
"time"
"github.com/mikespook/gearman-go/client"
)
@ -14,29 +13,29 @@ func main() {
// by implementing IdGenerator interface.
// client.IdGen = client.NewAutoIncId()
c, err := client.New("127.0.0.1:4730")
c, err := client.New("tcp4", "127.0.0.1:4730")
if err != nil {
log.Fatalln(err)
}
defer c.Close()
c.ErrHandler = func(e error) {
c.ErrorHandler = func(e error) {
log.Println(e)
}
echo := []byte("Hello\x00 world")
wg.Add(1)
echomsg, err := c.Echo(echo, time.Second)
echomsg, err := c.Echo(echo)
if err != nil {
log.Fatalln(err)
}
log.Println(string(echomsg))
wg.Done()
jobHandler := func(job *client.Job) {
jobHandler := func(job *client.Response) {
log.Printf("%s", job.Data)
wg.Done()
}
handle := c.Do("ToUpper", echo, client.JOB_NORMAL, jobHandler)
handle, err := c.Do("ToUpper", echo, client.JOB_NORMAL, jobHandler)
wg.Add(1)
status, err := c.Status(handle, time.Second)
status, err := c.Status(handle)
if err != nil {
log.Fatalln(err)
}

View File

@ -9,29 +9,25 @@ import (
"github.com/mikespook/gearman-go/worker"
)
func ToUpper(job *worker.Job) ([]byte, error) {
log.Printf("ToUpper: Handle=[%s]; UID=[%s], Data=[%s]\n",
job.Handle, job.UniqueId, job.Data)
data := []byte(strings.ToUpper(string(job.Data)))
func ToUpper(job worker.Job) ([]byte, error) {
log.Printf("ToUpper: Data=[%s]\n", job.Data())
data := []byte(strings.ToUpper(string(job.Data())))
return data, nil
}
func ToUpperDelay10(job *worker.Job) ([]byte, error) {
log.Printf("ToUpperDelay10: Handle=[%s]; UID=[%s], Data=[%s]\n",
job.Handle, job.UniqueId, job.Data)
func ToUpperDelay10(job worker.Job) ([]byte, error) {
log.Printf("ToUpper: Data=[%s]\n", job.Data())
time.Sleep(10 * time.Second)
data := []byte(strings.ToUpper(string(job.Data)))
data := []byte(strings.ToUpper(string(job.Data())))
return data, nil
}
func main() {
log.Println("Starting ...")
defer log.Println("Shutdown complete!")
w := worker.New(worker.Unlimited)
defer w.Close()
w.ErrHandler = func(e error) {
w.ErrorHandler = func(e error) {
log.Println(e)
if e == worker.ErrConnection {
proc, err := os.FindProcess(os.Getpid())
@ -43,12 +39,11 @@ func main() {
}
}
}
w.JobHandler = func(job *worker.Job) error {
log.Printf("H=%s, UID=%s, Data=%s, DataType=%d\n", job.Handle,
job.UniqueId, job.Data, job.DataType)
w.JobHandler = func(job worker.Job) error {
log.Printf("Data=%s\n", job.Data())
return nil
}
w.AddServer("127.0.0.1:4730")
w.AddServer("tcp4", "127.0.0.1:4730")
w.AddFunc("ToUpper", ToUpper, worker.Immediately)
w.AddFunc("ToUpperTimeOut5", ToUpperDelay10, 5)
w.AddFunc("ToUpperTimeOut20", ToUpperDelay10, 20)

View File

@ -12,20 +12,20 @@ import (
)
// Worker side job
type Response struct {
type InPack struct {
DataType uint32
Data []byte
Handle, UniqueId, Fn string
agentId string
a *agent
}
// Create a new job
func getResponse() (resp *Response) {
return &Response{}
func getInPack() (resp *InPack) {
return &InPack{}
}
// Decode job from byte slice
func decodeResponse(data []byte) (resp *Response, l int, err error) {
func decodeInPack(data []byte) (resp *InPack, l int, err error) {
if len(data) < MIN_PACKET_LEN { // valid package should not less 12 bytes
err = fmt.Errorf("Invalid data: %V", data)
return
@ -36,7 +36,7 @@ func decodeResponse(data []byte) (resp *Response, l int, err error) {
err = fmt.Errorf("Invalid data: %V", data)
return
}
resp = getResponse()
resp = getInPack()
resp.DataType = binary.BigEndian.Uint32(data[4:8])
switch resp.DataType {
case JOB_ASSIGN:

View File

@ -10,19 +10,19 @@ import (
)
// Worker side job
type request struct {
type OutPack struct {
DataType uint32
Data []byte
Handle, UniqueId, Fn string
}
func getRequest() (req *request) {
func getOutPack() (req *OutPack) {
// TODO pool
return &request{}
return &OutPack{}
}
// Encode a job to byte slice
func (req *request) Encode() (data []byte) {
func (req *OutPack) Encode() (data []byte) {
var l int
if req.DataType == WORK_FAIL {
l = len(req.Handle)

View File

@ -183,7 +183,6 @@ func (worker *Worker) Work() {
}
var resp *Response
for resp = range worker.in {
fmt.Println(resp)
go worker.dealResp(resp)
}
}