|
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110 |
- 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) 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 + 3)
- 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]))
- 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
- }
|