diff --git a/src/pkg/gearman/job.go b/src/pkg/gearman/job.go new file mode 100644 index 0000000..54d2088 --- /dev/null +++ b/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 +} + + + diff --git a/src/pkg/gearman/jobclient.go b/src/pkg/gearman/jobclient.go new file mode 100644 index 0000000..50e71e0 --- /dev/null +++ b/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 +}