Fix reschedule isuues (#22)

* remove rescheduling job while running worker stopped * Seperate jobDone, jobFailed, jobFailedWithException Support CanDOTimeout * remove dependency from db & fix some error * Server and worker restart support - set timeout for every running job - monitor all running job & remove if timeout expire * report to client that job failed with timeout exception * generic db call for all. * DB test fixed * fix * Merge Pull Request mikespook/gearman-go#75, Add sync lock to create job functions Signed-off-by: sadlil <[email protected]>
appscode · Mar 9, 2017 · 8ddd88c · 8ddd88c
1 parent e009f36
commit 8ddd88c
Show file tree

Hide file tree

Showing 12 changed files with 563 additions and 345 deletions.
diff --git a/client/client.go b/client/client.go
@@ -222,6 +222,8 @@ func (client *Client) do(funcname string, data []byte, flag rt.PT) (handle strin
 		return "", ErrLostConn
 	}
 	var result = make(chan handleOrError, 1)
+	client.Lock()
+	defer client.Unlock()
 	client.innerHandler.put("c", func(resp *Response) {
 		if resp.DataType == rt.PT_Error {
 			err = getError(resp.Data)

diff --git a/client/client_test.go b/client/client_test.go
@@ -1,6 +1,8 @@
 package client
 
 import (
+	"fmt"
+	"sync"
 	"testing"
 	"time"
 
@@ -48,7 +50,7 @@ func TestClientEcho(t *testing.T) {
 }
 
 func TestClientDoBg(t *testing.T) {
-	handle, err := client.DoBg("ToUpper", []byte("abcdef"), rt.JobNormal)
+	handle, err := client.DoBg("scheduledJobTest", []byte("abcdef"), rt.JobNormal)
 	if err != nil {
 		t.Error(err)
 		return
@@ -61,7 +63,7 @@ func TestClientDoBg(t *testing.T) {
 }
 
 func TestClientDoCron(t *testing.T) {
-	handle, err := client.DoCron("scheduledJobTest", "* * * * *", []byte("test data"))
+	handle, err := client.DoCron("scheduledJobTest", "* * * * 5", []byte("test data"))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -73,7 +75,7 @@ func TestClientDoCron(t *testing.T) {
 }
 
 func TestClientDoAt(t *testing.T) {
-	handle, err := client.DoAt("scheduledJobTest", time.Now().Add(10*time.Second).Unix(), []byte("test data"))
+	handle, err := client.DoAt("scheduledJobTest", time.Now().Add(20*time.Second).Unix(), []byte("test data"))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -85,17 +87,28 @@ func TestClientDoAt(t *testing.T) {
 }
 
 func TestClientDo(t *testing.T) {
+	var wg sync.WaitGroup = sync.WaitGroup{}
+	wg.Add(1)
 	jobHandler := func(job *Response) {
-		str := string(job.Data)
-		if str == "ABCDEF" {
-			t.Log(str)
-		} else {
-			t.Errorf("Invalid data: %s", job.Data)
+		switch job.DataType {
+		case rt.PT_WorkComplete:
+			t.Log("Work complete, handle ", job.Handle)
+			wg.Done()
+		case rt.PT_WorkException, rt.PT_WorkFail:
+			t.Log("Work fail, handle ", job.Handle, " cause: ", string(job.Data))
+			wg.Done()
+		case rt.PT_WorkData:
+			t.Logf("Work data: %+v", string(job.Data))
+		case rt.PT_WorkStatus:
+			status, err := job.Status()
+			if err != nil {
+				t.Error(err)
+			}
+			fmt.Println("Work status, num: %v, denom: %v", status.Numerator, status.Denominator)
 		}
-		return
 	}
-	handle, err := client.Do("ToUpper", []byte("abcdef"),
-		rt.JobLow, jobHandler)
+	handle, err := client.Do("scheduledJobTest", []byte("abcdef"),
+		rt.JobHigh, jobHandler)
 	if err != nil {
 		t.Error(err)
 		return
@@ -105,6 +118,8 @@ func TestClientDo(t *testing.T) {
 	} else {
 		t.Log(handle)
 	}
+	wg.Wait()
+
 }
 
 func TestClientStatus(t *testing.T) {

diff --git a/pkg/runtime/job.go b/pkg/runtime/job.go
@@ -9,7 +9,7 @@ const (
 	PRIORITY_HIGH = 1
 
 	JobPrefix       = "H:"
-	SchedJobPrefix  = "S:"
+	CronJobPrefix   = "S:"
 	EpochTimePrefix = "UTC-"
 )
 
@@ -22,7 +22,7 @@ type Job struct {
 	Denominator  int       `json:"denominator,omitempty"`
 	CreateAt     time.Time `json:"created_at,omitempty"`
 	ProcessAt    time.Time `json:"process_at,omitempty"`
-	TimeoutSec   int       `json:"timeout_sec,omitempty"`
+	TimeoutSec   int32     `json:"timeout_sec,omitempty"`
 	CreateBy     int64     `json:"created_by,omitempty"` //client sessionId
 	ProcessBy    int64     `json:"process_by,omitempty"` //worker sessionId
 	FuncName     string    `json:"function_name,omitempty"`
@@ -42,3 +42,19 @@ type CronJob struct {
 	SuccessfulRun int       `json:"successful_run,omitempty"`
 	FailedRun     int       `json:"failed_run,omitempty"`
 }
+
+func (c *Job) Key() string {
+	return c.Handle
+}
+
+func (c *Job) Prefix() string {
+	return JobPrefix
+}
+
+func (c *CronJob) Key() string {
+	return c.Handle
+}
+
+func (c *CronJob) Prefix() string {
+	return CronJobPrefix
+}
diff --git a/pkg/runtime/protocol.go b/pkg/runtime/protocol.go
@@ -82,8 +82,9 @@ const (
 	Req    = 5391697
 	ReqStr = "\x00REQ"
 	// \x00RES
-	Res    = 5391699
-	ResStr = "\x00RES"
+	Res            = 5391699
+	ResStr         = "\x00RES"
+	DefaultTimeout = 20 // 20 Seconds
 
 	HANDLE_SHAKE_HEADER_LENGTH = 12
 )