on job no longer in results, delete from matrix;; no backwards compatible
This commit is contained in:
59
main.go
59
main.go
@@ -9,6 +9,7 @@ import (
|
||||
"local/truckstop/config"
|
||||
"local/truckstop/logtr"
|
||||
"local/truckstop/message"
|
||||
"log"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"regexp"
|
||||
@@ -235,7 +236,7 @@ func _main() error {
|
||||
logtr.Errorf("failed _main: %v", err)
|
||||
}
|
||||
if config.Get().Once {
|
||||
time.Sleep(time.Second)
|
||||
time.Sleep(3 * time.Second)
|
||||
return err
|
||||
}
|
||||
if err != nil {
|
||||
@@ -267,6 +268,11 @@ func once() error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
logtr.Debugf("once: update dead jobs: %+v", alljobs)
|
||||
err = updateDeadJobs(alljobs)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
logtr.Debugf("once: all jobs: %+v", alljobs)
|
||||
newjobs, err := dropStaleJobs(alljobs)
|
||||
if err != nil {
|
||||
@@ -314,6 +320,45 @@ func getJobs() ([]broker.Job, error) {
|
||||
return jobs, nil
|
||||
}
|
||||
|
||||
type recordedJob struct {
|
||||
Job broker.Job
|
||||
SentNS int64
|
||||
MatrixID string
|
||||
}
|
||||
|
||||
func updateDeadJobs(jobs []broker.Job) error {
|
||||
db := config.Get().DB()
|
||||
list, err := db.List([]string{}, "sent_job_", "sent_job_}}")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
log.Printf("db.List() => %+v", list)
|
||||
for _, listEntry := range list {
|
||||
wouldBe := strings.TrimPrefix(listEntry, "sent_job_")
|
||||
found := false
|
||||
for i := range jobs {
|
||||
found = found || jobs[i].ID == wouldBe
|
||||
}
|
||||
logtr.Debugf("found job %s to be still alive==%v", wouldBe, found)
|
||||
if !found {
|
||||
logtr.Debugf("updating dead job %+v", listEntry)
|
||||
b, err := db.Get(listEntry)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var recorded recordedJob
|
||||
if err := json.Unmarshal(b, &recorded); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := message.NewMatrix().Update(recorded.MatrixID, recorded.Job.FormatMultilineTextDead()); err != nil {
|
||||
return err
|
||||
}
|
||||
db.Set(listEntry, nil)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func dropStaleJobs(jobs []broker.Job) ([]broker.Job, error) {
|
||||
db := config.Get().DB()
|
||||
for i := len(jobs) - 1; i >= 0; i-- {
|
||||
@@ -345,17 +390,17 @@ func sendJob(job broker.Job) (bool, error) {
|
||||
}
|
||||
func() {
|
||||
db := config.Get().DB()
|
||||
b, err := json.Marshal(map[string]interface{}{
|
||||
"Job": job,
|
||||
"SentTS": time.Now().Unix(),
|
||||
"MatrixID": id,
|
||||
b, err := json.Marshal(recordedJob{
|
||||
Job: job,
|
||||
SentNS: time.Now().UnixNano(),
|
||||
MatrixID: id,
|
||||
})
|
||||
if err != nil {
|
||||
logtr.Debugf("cannot marshal job+meta: %v", err)
|
||||
logtr.Errorf("failed to marshal recorded job: %v", err)
|
||||
return
|
||||
}
|
||||
if err := db.Set("sent_job_"+job.ID, b); err != nil {
|
||||
logtr.Debugf("cannot db.set job+meta: %v", err)
|
||||
logtr.Errorf("failed to set recorded job: %v", err)
|
||||
return
|
||||
}
|
||||
}()
|
||||
|
||||
Reference in New Issue
Block a user