Merge pull request #60 from ch3nnn/master

fix: producerNode.Delay func not wrap()

dq/consumernode.go

@@ -1,6 +1,7 @@
 package dq
 
 import (
+	"errors"
 	"time"
 
 	"github.com/beanstalkd/go-beanstalk"
@@ -59,14 +60,25 @@ func (c *consumerNode) consumeEvents(consume Consume) {
 		}
 
 		// the error can only be beanstalk.NameError or beanstalk.ConnError
-		switch cerr := err.(type) {
-		case beanstalk.ConnError:
-			switch cerr.Err {
-			case beanstalk.ErrTimeout:
+		var cerr beanstalk.ConnError
+		switch {
+		case errors.As(err, &cerr):
+			switch {
+			case errors.Is(cerr.Err, beanstalk.ErrTimeout):
 				// timeout error on timeout, just continue the loop
-			case beanstalk.ErrBadChar, beanstalk.ErrBadFormat, beanstalk.ErrBuried, beanstalk.ErrDeadline,
-				beanstalk.ErrDraining, beanstalk.ErrEmpty, beanstalk.ErrInternal, beanstalk.ErrJobTooBig,
-				beanstalk.ErrNoCRLF, beanstalk.ErrNotFound, beanstalk.ErrNotIgnored, beanstalk.ErrTooLong:
+			case
+				errors.Is(cerr.Err, beanstalk.ErrBadChar),
+				errors.Is(cerr.Err, beanstalk.ErrBadFormat),
+				errors.Is(cerr.Err, beanstalk.ErrBuried),
+				errors.Is(cerr.Err, beanstalk.ErrDeadline),
+				errors.Is(cerr.Err, beanstalk.ErrDraining),
+				errors.Is(cerr.Err, beanstalk.ErrEmpty),
+				errors.Is(cerr.Err, beanstalk.ErrInternal),
+				errors.Is(cerr.Err, beanstalk.ErrJobTooBig),
+				errors.Is(cerr.Err, beanstalk.ErrNoCRLF),
+				errors.Is(cerr.Err, beanstalk.ErrNotFound),
+				errors.Is(cerr.Err, beanstalk.ErrNotIgnored),
+				errors.Is(cerr.Err, beanstalk.ErrTooLong):
 				// won't reset
 				logx.Error(err)
 			default:

dq/producer.go

@@ -1,10 +1,8 @@
 package dq
 
 import (
-	"bytes"
 	"log"
 	"math/rand"
-	"strconv"
 	"strings"
 	"time"
 
@@ -56,9 +54,8 @@ func NewProducer(beanstalks []Beanstalk) Producer {
 }
 
 func (p *producerCluster) At(body []byte, at time.Time) (string, error) {
-	wrapped := p.wrap(body, at)
 	return p.insert(func(node Producer) (string, error) {
-		return node.At(wrapped, at)
+		return node.At(body, at)
 	})
 }
 
@@ -73,9 +70,8 @@ func (p *producerCluster) Close() error {
 }
 
 func (p *producerCluster) Delay(body []byte, delay time.Duration) (string, error) {
-	wrapped := p.wrap(body, time.Now().Add(delay))
 	return p.insert(func(node Producer) (string, error) {
-		return node.Delay(wrapped, delay)
+		return node.Delay(body, delay)
 	})
 }
 
@@ -156,11 +152,3 @@ func (p *producerCluster) insert(fn func(node Producer) (string, error)) (string
 
 	return "", be.Err()
 }
-
-func (p *producerCluster) wrap(body []byte, at time.Time) []byte {
-	var builder bytes.Buffer
-	builder.WriteString(strconv.FormatInt(at.UnixNano(), 10))
-	builder.WriteByte(timeSep)
-	builder.Write(body)
-	return builder.Bytes()
-}

dq/producernode.go

@@ -1,13 +1,15 @@
 package dq
 
 import (
+	"bytes"
 	"errors"
 	"fmt"
 	"strconv"
 	"strings"
 	"time"
 
 	"github.com/beanstalkd/go-beanstalk"
+	"github.com/zeromicro/go-zero/core/logx"
 )
 
 var ErrTimeBeforeNow = errors.New("can't schedule task to past time")
@@ -46,24 +48,38 @@ func (p *producerNode) Delay(body []byte, delay time.Duration) (string, error) {
 		return "", err
 	}
 
-	id, err := conn.Put(body, PriNormal, delay, defaultTimeToRun)
+	wrapped := p.wrap(body, time.Now().Add(delay))
+	id, err := conn.Put(wrapped, PriNormal, delay, defaultTimeToRun)
 	if err == nil {
 		return fmt.Sprintf("%s/%s/%d", p.endpoint, p.tube, id), nil
 	}
 
 	// the error can only be beanstalk.NameError or beanstalk.ConnError
 	// just return when the error is beanstalk.NameError, don't reset
-	switch cerr := err.(type) {
-	case beanstalk.ConnError:
-		switch cerr.Err {
-		case beanstalk.ErrBadChar, beanstalk.ErrBadFormat, beanstalk.ErrBuried, beanstalk.ErrDeadline,
-			beanstalk.ErrDraining, beanstalk.ErrEmpty, beanstalk.ErrInternal, beanstalk.ErrJobTooBig,
-			beanstalk.ErrNoCRLF, beanstalk.ErrNotFound, beanstalk.ErrNotIgnored, beanstalk.ErrTooLong:
+	var cerr beanstalk.ConnError
+	switch {
+	case errors.As(err, &cerr):
+		switch {
+		case
+			errors.Is(cerr.Err, beanstalk.ErrBadChar),
+			errors.Is(cerr.Err, beanstalk.ErrBadFormat),
+			errors.Is(cerr.Err, beanstalk.ErrBuried),
+			errors.Is(cerr.Err, beanstalk.ErrDeadline),
+			errors.Is(cerr.Err, beanstalk.ErrDraining),
+			errors.Is(cerr.Err, beanstalk.ErrEmpty),
+			errors.Is(cerr.Err, beanstalk.ErrInternal),
+			errors.Is(cerr.Err, beanstalk.ErrJobTooBig),
+			errors.Is(cerr.Err, beanstalk.ErrNoCRLF),
+			errors.Is(cerr.Err, beanstalk.ErrNotFound),
+			errors.Is(cerr.Err, beanstalk.ErrNotIgnored),
+			errors.Is(cerr.Err, beanstalk.ErrTooLong):
 			// won't reset
 		default:
 			// beanstalk.ErrOOM, beanstalk.ErrTimeout, beanstalk.ErrUnknown and other errors
 			p.conn.reset()
 		}
+	default:
+		logx.Error(err)
 	}
 
 	return "", err
@@ -96,3 +112,11 @@ func (p *producerNode) Revoke(jointId string) error {
 	// if not in this beanstalk, ignore
 	return nil
 }
+
+func (p *producerNode) wrap(body []byte, at time.Time) []byte {
+	var builder bytes.Buffer
+	builder.WriteString(strconv.FormatInt(at.UnixNano(), 10))
+	builder.WriteByte(timeSep)
+	builder.Write(body)
+	return builder.Bytes()
+}

example/dq/producer/cluster.go

@@ -0,0 +1,28 @@
+package main
+
+import (
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/zeromicro/go-queue/dq"
+)
+
+func main() {
+	producer := dq.NewProducer([]dq.Beanstalk{
+		{
+			Endpoint: "localhost:11300",
+			Tube:     "tube",
+		},
+		{
+			Endpoint: "localhost:11300",
+			Tube:     "tube",
+		},
+	})
+	for i := 1000; i < 1005; i++ {
+		_, err := producer.Delay([]byte(strconv.Itoa(i)), time.Second*5)
+		if err != nil {
+			fmt.Println(err)
+		}
+	}
+}

example/dq/producer/producer.go

@@ -9,16 +9,8 @@ import (
 )
 
 func main() {
-	producer := dq.NewProducer([]dq.Beanstalk{
-		{
-			Endpoint: "localhost:11300",
-			Tube:     "tube",
-		},
-		{
-			Endpoint: "localhost:11300",
-			Tube:     "tube",
-		},
-	})
+	producer := dq.NewProducerNode("localhost:11300", "tube")
+
 	for i := 1000; i < 1005; i++ {
 		_, err := producer.Delay([]byte(strconv.Itoa(i)), time.Second*5)
 		if err != nil {