Job server can echo sth. to worker now. Added files.

src/pkg/gearman/job.go

@@ -0,0 +1,75 @@
+package gearman
+
+import (
+    "os"
+)
+
+const (
+    // \x00REQ
+    REQ = 5391697
+    // \x00RES
+    RES = 5391699
+    ECHO = "\x00REQ\x00\x00\x00\x10\x00\x00\x00\x00"
+    ECHO_REQ = 16
+    ECHO_RES = 17
+)
+
+type Job struct {
+    client *JobClient
+    Data []byte
+    MagicCode, DataType uint32
+}
+
+func ByteToUint32(buf [4]byte) uint32 {
+    return uint32(buf[0]) << 24 +
+        uint32(buf[1]) << 16 +
+        uint32(buf[2]) << 8 +
+        uint32(buf[3])
+}
+
+func Uint32ToByte(i uint32) (data [4]byte) {
+    data[0] = byte((i >> 24) & 0xff)
+    data[1] = byte((i >> 16) & 0xff)
+    data[2] = byte((i >> 8) & 0xff)
+    data[3] = byte(i & 0xff)
+    return
+}
+
+func NewJob(server *JobClient, magiccode, datatype uint32, data []byte) (job *Job) {
+    return &Job{client: server,
+        MagicCode:magiccode,
+        DataType: datatype,
+        Data:data}
+}
+
+func DecodeJob(server *JobClient, data []byte) (job *Job, err os.Error) {
+    if len(data) < 12 {
+        return nil, os.NewError("Data length is too small.")
+    }
+    datatype := ByteToUint32([4]byte{data[4], data[5], data[6], data[7]})
+    l := ByteToUint32([4]byte{data[8], data[9], data[10], data[11]})
+    if len(data[12:]) != int(l) {
+         return nil, os.NewError("Invalid data length.")
+    }
+    switch(ByteToUint32([4]byte{data[4], data[5], data[6], data[7]})) {
+        case ECHO_RES:
+            data = data[12:]
+    }
+    return NewJob(server, REQ, datatype, data), err
+}
+
+func (job *Job) Encode() (data []byte) {
+    magiccode := Uint32ToByte(job.MagicCode)
+    datatype := Uint32ToByte(job.DataType)
+    l := len(job.Data)
+    datalength := Uint32ToByte(uint32(l))
+    data = make([]byte, 12 + l)
+    copy(data[:4], magiccode[:])
+    copy(data[4:8], datatype[:])
+    copy(data[8:12], datalength[:])
+    copy(data[12:], job.Data)
+    return
+}
+
+
+

src/pkg/gearman/jobclient.go

@@ -0,0 +1,73 @@
+package gearman
+
+import (
+    "net"
+    "os"
+    "log"
+)
+
+type JobClient struct {
+    conn net.Conn
+    incoming chan *Job
+    running bool
+}
+
+func NewJobClient(addr string, incoming chan *Job) (jobclient *JobClient, err os.Error) {
+    conn, err := net.Dial(TCP, addr)
+    if err != nil {
+        return nil, err
+    }
+    jobclient = &JobClient{conn: conn, incoming: incoming, running:true}
+    return jobclient, err
+}
+
+func (server *JobClient) Work() (err os.Error) {
+    for server.running {
+        var rel []byte
+        for {
+            buf := make([]byte, 2048)
+            n, err := server.conn.Read(buf)
+            if err != nil {
+                if err == os.EOF && n == 0 {
+                    break
+                }
+                return err
+            }
+            rel = append(rel, buf[0: n] ...)
+            break
+        }
+        job, err := DecodeJob(server, rel)
+        if err != nil {
+            log.Println(err)
+        } else {
+            server.incoming <- job
+        }
+    }
+    return
+}
+
+func (server *JobClient) WriteJob(job * Job) (err os.Error) {
+    return server.Write(job.Encode())
+}
+
+func (server *JobClient) Echo(str []byte) (err os.Error) {
+    job := NewJob(server, REQ, ECHO_REQ, []byte(str))
+    return server.Write(job.Encode());
+}
+
+func (server *JobClient) Write(buf []byte) (err os.Error) {
+    var n int
+    for i := 0; i < len(buf); i += n {
+        n, err = server.conn.Write(buf[i:])
+        if err != nil {
+            return err
+        }
+    }
+    return
+}
+
+func (server *JobClient) Close() (err os.Error) {
+    server.running = false
+    err = server.conn.Close()
+    return
+}