gearman-go/worker/inpack.go

package worker

import (
	"bytes"
	"encoding/binary"
	"fmt"
	"strconv"
)

// Worker side job
type inPack struct {
	dataType             uint32
	data                 []byte
	handle, uniqueId, fn string
	a                    *agent
}

// Create a new job
func getInPack() *inPack {
	return &inPack{}
}

func (inpack *inPack) Data() []byte {
	return inpack.data
}

func (inpack *inPack) Fn() string {
	return inpack.fn
}

func (inpack *inPack) Handle() string {
	return inpack.handle
}

func (inpack *inPack) UniqueId() string {
	return inpack.uniqueId
}

func (inpack *inPack) Err() error {
	if inpack.dataType == dtError {
		return getError(inpack.data)
	}
	return nil
}

// Send some datas to client.
// Using this in a job's executing.
func (inpack *inPack) SendData(data []byte) {
	outpack := getOutPack()
	outpack.dataType = dtWorkData
	hl := len(inpack.handle)
	l := hl + len(data) + 1
	outpack.data = getBuffer(l)
	copy(outpack.data, []byte(inpack.handle))
	copy(outpack.data[hl+1:], data)
	inpack.a.write(outpack)
}

func (inpack *inPack) SendWarning(data []byte) {
	outpack := getOutPack()
	outpack.dataType = dtWorkWarning
	hl := len(inpack.handle)
	l := hl + len(data) + 1
	outpack.data = getBuffer(l)
	copy(outpack.data, []byte(inpack.handle))
	copy(outpack.data[hl+1:], data)
	inpack.a.write(outpack)
}

// Update status.
// Tall client how many percent job has been executed.
func (inpack *inPack) UpdateStatus(numerator, denominator int) {
	n := []byte(strconv.Itoa(numerator))
	d := []byte(strconv.Itoa(denominator))
	outpack := getOutPack()
	outpack.dataType = dtWorkStatus
	hl := len(inpack.handle)
	nl := len(n)
	dl := len(d)
	outpack.data = getBuffer(hl + nl + dl + 2)
	copy(outpack.data, []byte(inpack.handle))
	copy(outpack.data[hl+1:], n)
	copy(outpack.data[hl+nl+2:], d)
	inpack.a.write(outpack)
}

// Decode job from byte slice
func decodeInPack(data []byte) (inpack *inPack, l int, err error) {
	if len(data) < minPacketLength { // valid package should not less 12 bytes
		err = fmt.Errorf("Invalid data: %v", data)
		return
	}
	dl := int(binary.BigEndian.Uint32(data[8:12]))
	if len(data) < (dl + minPacketLength) {
		err = fmt.Errorf("Not enough data: %v", data)
		return
	}
	dt := data[minPacketLength : dl+minPacketLength]
	if len(dt) != int(dl) { // length not equal
		err = fmt.Errorf("Invalid data: %v", data)
		return
	}
	inpack = getInPack()
	inpack.dataType = binary.BigEndian.Uint32(data[4:8])
	switch inpack.dataType {
	case dtJobAssign:
		s := bytes.SplitN(dt, []byte{'\x00'}, 3)
		if len(s) == 3 {
			inpack.handle = string(s[0])
			inpack.fn = string(s[1])
			inpack.data = s[2]
		}
	case dtJobAssignUniq:
		s := bytes.SplitN(dt, []byte{'\x00'}, 4)
		if len(s) == 4 {
			inpack.handle = string(s[0])
			inpack.fn = string(s[1])
			inpack.uniqueId = string(s[2])
			inpack.data = s[3]
		}
	default:
		inpack.data = dt
	}
	l = dl + minPacketLength
	return
}