change integration for new sdk

Author: changsongl

go.mod

@@ -4,7 +4,7 @@ go 1.14
 
 require (
 	github.com/agiledragon/gomonkey v2.0.2+incompatible
-	github.com/changsongl/delay-queue-client v0.0.3
+	github.com/changsongl/delay-queue-client v0.1.0-beta.0.20210727025911-85af22aa66a5
 	github.com/gin-gonic/gin v1.6.3
 	github.com/go-playground/validator/v10 v10.4.1 // indirect
 	github.com/go-redis/redis/v8 v8.4.4

go.sum

@@ -35,6 +35,8 @@ github.com/cespare/xxhash/v2 v2.1.1 h1:6MnRN8NT7+YBpUIWxHtefFZOKTAPgGjpQSxqLNn0+
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/changsongl/delay-queue-client v0.0.3 h1:Vz1b1g3WXbjQVs/FdufWKWCpM5HdEsCC3u5RyYTuJa4=
 github.com/changsongl/delay-queue-client v0.0.3/go.mod h1:H0o8Om/fu3QiyPLPrUop298xEUXLmlv+80wPH98v5iI=
+github.com/changsongl/delay-queue-client v0.1.0-beta.0.20210727025911-85af22aa66a5 h1:HhAIy5lINlXTz34orgrNoqUagt0yOoQorYNAWRT8C7M=
+github.com/changsongl/delay-queue-client v0.1.0-beta.0.20210727025911-85af22aa66a5/go.mod h1:H0o8Om/fu3QiyPLPrUop298xEUXLmlv+80wPH98v5iI=
 github.com/clbanning/x2j v0.0.0-20191024224557-825249438eec/go.mod h1:jMjuTZXRI4dUb/I5gc9Hdhagfvm9+RyrPryS/auMzxE=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/cockroachdb/datadriven v0.0.0-20190809214429-80d97fb3cbaa/go.mod h1:zn76sxSg3SzpJ0PPJaLDCu+Bu0Lg3sKTORVIj19EIF8=

test/integration/integration_test.go

@@ -31,7 +31,7 @@ func TestDelayQueueAddAndRemove(t *testing.T) {
 	for i := 0; i < Jobs; i++ {
 		delayTime := rand.Intn(DelayTimeSeconds)
 		id := fmt.Sprintf("test-%d", i)
-		j, err := job.New(topic, id, job.JobDelayOption(time.Duration(delayTime)*time.Second))
+		j, err := job.New(topic, id, job.DelayOption(time.Duration(delayTime)*time.Second))
 		require.NoError(t, err)
 
 		err = AddJobRecord(key, id)
@@ -49,7 +49,7 @@ func TestDelayQueueAddAndRemove(t *testing.T) {
 		c := consumer.New(cli, topic, consumer.WorkerNumOption(1))
 		ch := c.Consume()
 		for jobMsg := range ch {
-			id := jobMsg.GetId()
+			id := jobMsg.GetID()
 			err := DeleteJobRecord(key, id)
 			require.NoError(t, err)
 
@@ -78,7 +78,7 @@ func TestDelayQueueTTR(t *testing.T) {
 	t.Parallel()
 
 	topic, id := "TestDelayQueueTTR-topic", "000"
-	j, err := job.New(topic, id, job.JobDelayOption(10*time.Second), job.JobTTROption(5*time.Second))
+	j, err := job.New(topic, id, job.DelayOption(10*time.Second), job.TTROption(5*time.Second))
 	require.NoError(t, err)
 
 	cli := client.NewClient(DelayQueueAddr)
@@ -94,7 +94,7 @@ func TestDelayQueueTTR(t *testing.T) {
 		c := consumer.New(cli, topic, consumer.WorkerNumOption(2))
 		ch := c.Consume()
 		for jobMsg := range ch {
-			jobID := jobMsg.GetId()
+			jobID := jobMsg.GetID()
 			t.Logf("Receive job(id: %s): %d", jobID, time.Now().Unix())
 			if id == jobID {
 				v := atomic.LoadInt64(&num)
@@ -109,42 +109,41 @@ func TestDelayQueueTTR(t *testing.T) {
 	require.LessOrEqual(t, int64(4), num, "retry time should be equal")
 }
 
-// TODO: test block
-// Testing ttr, consume but don't finish or delete.
-// Message should be consume again.
-//func TestDelayQueueBlockPop(t *testing.T) {
-//	t.Parallel()
-//
-//	topic, id := "TestDelayQueueBlockPop-topic", "111"
-//	j, err := job.New(topic, id, job.JobDelayOption(0*time.Second))
-//	require.NoError(t, err)
-//
-//	blockTime := 5 * time.Second
-//
-//	cli := client.NewClient(DelayQueueAddr)
-//
-//	var totalTime time.Duration = 0
-//	go func() {
-//		// consume jobs
-//		c := consumer.New(cli, topic, consumer.WorkerNumOption(1))
-//		ch := c.Consume()
-//		startTime := time.Now()
-//		for jobMsg := range ch {
-//			jobID := jobMsg.GetId()
-//			t.Logf("Receive job(id: %s): %d", jobID, time.Now().Unix())
-//			if id == jobID {
-//				totalTime += time.Since(startTime)
-//			}
-//		}
-//	}()
-//
-//	time.Sleep(blockTime - 3*time.Second)
-//	t.Logf("Add job: %d", time.Now().Unix())
-//	err = cli.AddJob(j)
-//	require.NoError(t, err)
-//
-//	time.Sleep(blockTime)
-//	t.Log("total-time", totalTime)
-//	require.Greater(t, totalTime, time.Duration(0))
-//	require.LessOrEqual(t, totalTime, blockTime)
-//}
+//Testing ttr, consume but don't finish or delete.
+//Message should be consume again.
+func TestDelayQueueBlockPop(t *testing.T) {
+	t.Parallel()
+
+	topic, id := "TestDelayQueueBlockPop-topic", "111"
+	j, err := job.New(topic, id, job.DelayOption(0*time.Second))
+	require.NoError(t, err)
+
+	blockTime := 5 * time.Second
+
+	cli := client.NewClient(DelayQueueAddr)
+
+	var totalTime time.Duration = 0
+	go func() {
+		// consume jobs
+		c := consumer.New(cli, topic, consumer.WorkerNumOption(1), consumer.PopTimeoutOption(blockTime))
+		ch := c.Consume()
+		startTime := time.Now()
+		for jobMsg := range ch {
+			jobID := jobMsg.GetID()
+			t.Logf("Receive job(id: %s): %d", jobID, time.Now().Unix())
+			if id == jobID {
+				totalTime += time.Since(startTime)
+			}
+		}
+	}()
+
+	time.Sleep(blockTime - 3*time.Second)
+	t.Logf("Add job: %d", time.Now().Unix())
+	err = cli.AddJob(j)
+	require.NoError(t, err)
+
+	time.Sleep(blockTime)
+	t.Log("total-time", totalTime)
+	require.Greater(t, totalTime, time.Duration(0))
+	require.LessOrEqual(t, totalTime, blockTime)
+}