chtwrs/chirpnest
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

update load obj job

Browse Source
This commit is contained in:
shoopea 2020-01-21 15:46:27 +08:00
parent d52e58493a
commit e99839e67a
1 changed files with 16 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
obj.go
View File

@ -1085,20 +1085,21 @@ func loadObjJob() error {
timeout time.Time
user int64
zpayload []byte
zipped int
)
muxObjJob.Lock()
cacheObjJob = make(map[int64]Job)
muxObjJob.Unlock()
jobs, err := db.Query(`SELECT o.id, o.obj_sub_type_id, oj.trigger_id, oj.timeout, oj.user_id, oj.payload FROM obj o, obj_job oj WHERE o.id = oj.obj_id;;`)
jobs, err := db.Query(`SELECT o.id, o.obj_sub_type_id, oj.trigger_id, oj.timeout, oj.user_id, oj.zipped, oj.payload FROM obj o, obj_job oj WHERE o.id = oj.obj_id;;`)
if err != nil {
return err
}
defer jobs.Close()
for jobs.Next() {
err = jobs.Scan(&id, &type_id, &trigger, &timeout, &user, &zpayload)
err = jobs.Scan(&id, &type_id, &trigger, &timeout, &user, &zipped, &zpayload)
if err != nil {
return err
}
@ -1109,16 +1110,20 @@ func loadObjJob() error {
j.Timeout = timeout
j.UserID64 = user
zb := bytes.NewReader(zpayload)
zr, err := zlib.NewReader(zb)
if err != nil {
logOnError(err, "loadObjJob : zlib.NewReader")
continue
if zipped > 0 {
zb := bytes.NewReader(zpayload)
zr, err := zlib.NewReader(zb)
if err != nil {
logOnError(err, "loadObjJob : zlib.NewReader")
continue
}
b := new(bytes.Buffer)
b.ReadFrom(zr)
payload := b.Bytes()
j.Payload = payload
} else {
j.Payload = zpayload
}
b := new(bytes.Buffer)
b.ReadFrom(zr)
payload := b.Bytes()
j.Payload = payload
muxObjJob.Lock()
cacheObjJob[id] = *j