MKP-329 Sync changes back to upstream

README.md

@@ -14,12 +14,12 @@ There are benefits of using Que to route jobs:
 ## Install
 
 ```bash
-go get github.com/tnclong/go-que
+go get github.com/theplant/go-que
 ```
 
 ## Doc
 
-https://godoc.org/github.com/tnclong/go-que
+https://godoc.org/github.com/theplant/go-que
 
 ## Quickstart
 

bm/go.mod

@@ -1,11 +1,11 @@
-module github.com/tnclong/go-que/bm
+module github.com/theplant/go-que/bm
 
 go 1.13
 
-replace github.com/tnclong/go-que => ../
+replace github.com/theplant/go-que => ../
 
 require (
 	github.com/lib/pq v1.3.0
-	github.com/tnclong/go-que v0.0.0-00010101000000-000000000000
+	github.com/theplant/go-que v0.0.0-00010101000000-000000000000
 	gopkg.in/yaml.v2 v2.3.0
 )

bm/main.go

@@ -18,8 +18,8 @@ import (
 	"time"
 
 	_ "github.com/lib/pq"
-	"github.com/tnclong/go-que"
-	"github.com/tnclong/go-que/pg"
+	"github.com/theplant/go-que"
+	"github.com/theplant/go-que/pg"
 	"gopkg.in/yaml.v2"
 )
 

example_test.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/tnclong/go-que"
+	"github.com/theplant/go-que"
 )
 
 func ExampleWorker() {

go.mod

@@ -1,4 +1,4 @@
-module github.com/tnclong/go-que
+module github.com/theplant/go-que
 
 go 1.13
 

pg/job.go

@@ -5,7 +5,7 @@ import (
 	"database/sql"
 	"time"
 
-	"github.com/tnclong/go-que"
+	"github.com/theplant/go-que"
 )
 
 type job struct {
@@ -51,13 +51,15 @@ func (j *job) In(tx *sql.Tx) {
 }
 
 const doneJob = `UPDATE goque_jobs
-SET done_at = now()
-WHERE id = $1::bigint`
+SET done_at = now(),
+    result = $1::jsonb
+WHERE id = $2::bigint`
 
 const doneUniqueIDJob = `UPDATE goque_jobs
 SET done_at = now(),
+    result = $1::jsonb,
     unique_id = null
-WHERE id = $1::bigint`
+WHERE id = $2::bigint`
 
 func (j *job) Done(ctx context.Context) error {
 	var execSQL string
@@ -66,7 +68,12 @@ func (j *job) Done(ctx context.Context) error {
 	} else {
 		execSQL = doneJob
 	}
-	_, err := j.exec(j.tx)(ctx, execSQL, j.id)
+	resultStr := "{}"
+	if result, ok := ctx.Value("result").(string); ok {
+		resultStr = result
+	}
+
+	_, err := j.exec(j.tx)(ctx, execSQL, resultStr, j.id)
 	return err
 }
 

pg/mutex.go

@@ -12,7 +12,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/tnclong/go-que"
+	"github.com/theplant/go-que"
 )
 
 type mutex struct {
@@ -214,18 +214,20 @@ func (m *mutex) lockInMux(ctx context.Context, conn *sql.Conn, queue string, cou
 	if err != nil {
 		return nil, nil, err
 	}
-	var doneAt, exipredAt sql.NullTime
+	var createdAt, doneAt, exipredAt sql.NullTime
 	var locked bool
 	var remaining int
+	var pushNotification string
+	var result []byte
 	defer rows.Close()
 	for rows.Next() {
 		var jb job
 		rp := (*jsonRetryPolicy)(&jb.plan.RetryPolicy)
 		var uniqueID sql.NullString
 		err = rows.Scan(
-			&jb.id, &jb.plan.Queue, &jb.plan.Args, &jb.plan.RunAt, rp, &doneAt, &exipredAt,
+			&jb.id, &jb.plan.Queue, &jb.plan.Args, &createdAt, &jb.plan.RunAt, rp, &doneAt, &exipredAt, &pushNotification,
 			&jb.retryCount, &jb.lastErrMsg, &jb.lastErrStack,
-			&uniqueID, &jb.plan.UniqueLifecycle,
+			&uniqueID, &jb.plan.UniqueLifecycle, &result,
 			&locked, &remaining,
 		)
 		if err != nil {

pg/queue.go

@@ -7,7 +7,7 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/tnclong/go-que"
+	"github.com/theplant/go-que"
 )
 
 func New(db *sql.DB) (que.Queue, error) {

pg/sql.go

@@ -12,11 +12,12 @@ create table if not exists goque_jobs
     args                 jsonb                    default '[]'::jsonb not null
         constraint valid_args
             check (jsonb_typeof(args) = 'array'::text),
-
+    created_at           timestamp with time zone default now()       not null,
     run_at               timestamp with time zone default now()       not null,
     retry_policy         jsonb                                        not null,
     done_at              timestamp with time zone,
     expired_at           timestamp with time zone,
+    push_notification    text default '',
 
     retry_count          integer                  default 0           not null,
     last_err_msg   text,
@@ -27,6 +28,10 @@ create table if not exists goque_jobs
         constraint valid_unique_lifecycle 
             check(unique_lifecycle>=0 AND unique_lifecycle<=3),
 
+    result               jsonb                    default '{}'::jsonb not null
+            constraint result
+                check (jsonb_typeof(result) = 'object'::text),
+
     constraint err_length
         check ((char_length(last_err_msg) <= 512) AND (char_length(last_err_stack) <= 8192))
 );

queue_test.go

@@ -17,8 +17,8 @@ import (
 	"time"
 
 	_ "github.com/lib/pq"
-	"github.com/tnclong/go-que"
-	"github.com/tnclong/go-que/pg"
+	"github.com/theplant/go-que"
+	"github.com/theplant/go-que/pg"
 )
 
 func TestEnqueueLockUnlock(t *testing.T) {

scheduler/provider_test.go

@@ -6,7 +6,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/tnclong/go-que"
+	"github.com/theplant/go-que"
 )
 
 var wantSchedule = Schedule{

scheduler/schedule.go

@@ -4,7 +4,7 @@ import (
 	"errors"
 
 	"github.com/robfig/cron/v3"
-	"github.com/tnclong/go-que"
+	"github.com/theplant/go-que"
 )
 
 // Schedule is a set of named items.

scheduler/scheduler.go

@@ -8,7 +8,7 @@ import (
 	"log"
 	"time"
 
-	"github.com/tnclong/go-que"
+	"github.com/theplant/go-que"
 )
 
 // Scheduler enqueue jobs according to provided schedule.

scheduler/scheduler_test.go

@@ -10,8 +10,8 @@ import (
 
 	"github.com/golang/mock/gomock"
 	_ "github.com/lib/pq"
-	"github.com/tnclong/go-que"
-	"github.com/tnclong/go-que/mock"
+	"github.com/theplant/go-que"
+	"github.com/theplant/go-que/mock"
 )
 
 func TestDecodeArgs(t *testing.T) {

stack_test.go

@@ -37,9 +37,9 @@ func TestStack(t *testing.T) {
 	t.Log(stack)
 
 	stacks := []string{
-		"github.com/tnclong/go-que/stack_test.go:36",
-		"github.com/tnclong/go-que/stack_test.go:17",
-		"github.com/tnclong/go-que/stack_test.go:31",
+		"github.com/theplant/go-que/stack_test.go:36",
+		"github.com/theplant/go-que/stack_test.go:17",
+		"github.com/theplant/go-que/stack_test.go:31",
 		"tnclong/go-que/stack_test.go:25",
 		"tnclong/go-que/stack_test.go:14",
 	}