feat: simple delay queue

Change-Id: I8b51bd35c4860843a592a6ef74b15e31350861d1

Author: yanglong.2017

.gitignore

@@ -12,3 +12,4 @@
 
 # Project-local glide cache, RE: https://github.com/Masterminds/glide/issues/736
 .glide/
+/later

README.md

@@ -0,0 +1,2 @@
+## later
+later is a redis base delay queue .

main.go

@@ -0,0 +1,31 @@
+package main
+
+import (
+	"flag"
+
+	"github.com/btfak/later/queue"
+	log "github.com/sirupsen/logrus"
+)
+
+var (
+	redisURL = flag.String("redis", "redis://127.0.0.1:6379/0", "redis address")
+	address  = flag.String("address", ":8080", "serve listen address")
+)
+
+func init() {
+	log.SetFormatter(&log.TextFormatter{FullTimestamp: true})
+	flag.Parse()
+}
+
+func main() {
+	err := queue.InitRedis(*redisURL)
+	if err != nil {
+		log.Fatal(err)
+	}
+	queue.RunWorker()
+	log.Infof("server listen on :%v", *address)
+	err = queue.ListenAndServe(*address)
+	if err != nil {
+		log.Fatal(err)
+	}
+}

queue/callback.go

@@ -0,0 +1,69 @@
+package queue
+
+import (
+	"bytes"
+	"encoding/json"
+	"io/ioutil"
+	"net/http"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+)
+
+var HttpClient = &http.Client{
+	Timeout: time.Second * 3,
+	Transport: &http.Transport{
+		MaxIdleConnsPerHost: 10,
+		MaxIdleConns:        1024,
+		IdleConnTimeout:     time.Minute * 5,
+	},
+}
+
+type callbackRequest struct {
+	ID      string `json:"id"`
+	Topic   string `json:"topic"`
+	Content string `json:"content"`
+}
+
+type callbackResponse struct {
+	Code int `json:"code"`
+}
+
+const (
+	CodeSuccess        = 100
+	CodeTooManyRequest = 101
+)
+
+func post(task *Task) (int, error) {
+	request := callbackRequest{
+		ID:      task.ID,
+		Topic:   task.Topic,
+		Content: task.Content,
+	}
+	data, err := json.Marshal(request)
+	if err != nil {
+		log.WithError(err).Error("json marshal fail")
+		return 0, err
+	}
+
+	content := bytes.NewBuffer(data)
+	resp, err := HttpClient.Post(task.Callback, "application/json", content)
+	if err != nil {
+		log.WithError(err).Error("http post fail")
+		return 0, err
+	}
+	defer resp.Body.Close()
+
+	result, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		log.WithError(err).Error("io read from backend fail")
+		return 0, err
+	}
+	var response callbackResponse
+	err = json.Unmarshal(result, &response)
+	if err != nil {
+		log.WithError(err).Error("json unmarshal fail")
+		return 0, err
+	}
+	return response.Code, nil
+}

queue/config.go

@@ -0,0 +1,22 @@
+package queue
+
+import (
+	"time"
+)
+
+var (
+	RedisConnectTimeout  = 50 * time.Millisecond
+	RedisReadTimeout     = 50 * time.Millisecond
+	RedisWriteTimeout    = 100 * time.Millisecond
+	RedisPoolMaxIdle     = 200
+	RedisPoolIdleTimeout = 3 * time.Minute
+)
+
+var (
+	TaskTTL             = 24 * 3600
+	ZrangeOffset        = 20
+	DelayWorkerInterval = 100 * time.Millisecond
+	UnackWorkerInterval = 1000 * time.Millisecond
+	ErrorWorkerInterval = 1000 * time.Millisecond
+	RetryInterval       = 10 //second
+)

queue/hack.go

@@ -0,0 +1,25 @@
+package queue
+
+import (
+	"reflect"
+	"unsafe"
+)
+
+// No copy to change slice to string, use your own risk
+func String(b []byte) (s string) {
+	pbytes := (*reflect.SliceHeader)(unsafe.Pointer(&b))
+	pstring := (*reflect.StringHeader)(unsafe.Pointer(&s))
+	pstring.Data = pbytes.Data
+	pstring.Len = pbytes.Len
+	return
+}
+
+// No copy to change string to slice, use your own risk
+func Slice(s string) (b []byte) {
+	pbytes := (*reflect.SliceHeader)(unsafe.Pointer(&b))
+	pstring := (*reflect.StringHeader)(unsafe.Pointer(&s))
+	pbytes.Data = pstring.Data
+	pbytes.Len = pstring.Len
+	pbytes.Cap = pstring.Len
+	return
+}

queue/redis.go

@@ -0,0 +1,22 @@
+package queue
+
+import (
+	"github.com/garyburd/redigo/redis"
+)
+
+var pool *redis.Pool
+
+func InitRedis(address string) error {
+	dial := func() (redis.Conn, error) {
+		return redis.DialURL(address,
+			redis.DialConnectTimeout(RedisConnectTimeout),
+			redis.DialReadTimeout(RedisReadTimeout),
+			redis.DialWriteTimeout(RedisWriteTimeout))
+	}
+	pool = &redis.Pool{
+		MaxIdle:     RedisPoolMaxIdle,
+		IdleTimeout: RedisPoolIdleTimeout,
+		Dial:        dial,
+	}
+	return nil
+}

queue/server.go

@@ -0,0 +1,74 @@
+package queue
+
+import (
+	"encoding/json"
+	"io/ioutil"
+	"net/http"
+	"time"
+
+	"github.com/pborman/uuid"
+	log "github.com/sirupsen/logrus"
+)
+
+type CreateTaskRequest struct {
+	// Topic use to classify tasks
+	Topic string `json:"topic"`
+	// Delay is the number of seconds that should elapse before the task execute
+	Delay int64 `json:"delay"`
+	// Retry is max deliver retry times
+	Retry int `json:"retry"`
+	// Callback is the deliver address
+	Callback string `json:"callback"`
+	// Content is the task content to deliver
+	Content string `json:"content"`
+}
+
+type CreateTaskResponse struct {
+	ID string `json:"id"`
+}
+
+func ListenAndServe(addr string) error {
+	http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPost {
+			w.WriteHeader(400)
+			return
+		}
+		var request CreateTaskRequest
+		data, err := ioutil.ReadAll(r.Body)
+		if err != nil {
+			log.WithError(err).Error("io read from frontend fail")
+			w.WriteHeader(500)
+			return
+		}
+		err = json.Unmarshal(data, &request)
+		if err != nil {
+			log.WithError(err).Error("json unmarshal fail")
+			w.WriteHeader(400)
+			return
+		}
+		task := &Task{
+			ID:          uuid.New(),
+			Topic:       request.Topic,
+			ExecuteTime: time.Now().Unix() + request.Delay,
+			MaxRetry:    request.Retry,
+			Callback:    request.Callback,
+			Content:     request.Content,
+			CreatTime:   time.Now().Unix(),
+		}
+		err = createTask(task)
+		if err != nil {
+			log.WithError(err).Error("create task fail")
+			w.WriteHeader(500)
+			return
+		}
+		response := CreateTaskResponse{ID: task.ID}
+		respData, err := json.Marshal(response)
+		if err != nil {
+			log.WithError(err).Error("json marshal fail")
+			w.WriteHeader(500)
+			return
+		}
+		w.Write(respData)
+	})
+	return http.ListenAndServe(addr, nil)
+}

queue/task.go

@@ -0,0 +1,138 @@
+package queue
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/garyburd/redigo/redis"
+)
+
+// Task is the task to execute
+type Task struct {
+	// ID is a global unique id
+	ID string
+	// Topic use to classify tasks
+	Topic string
+	// ExecuteTime is the time to deliver
+	ExecuteTime int64
+	// MaxRetry is max deliver retry times
+	MaxRetry int
+	//HasRetry is the current retry times
+	HasRetry int
+	// Callback is the deliver address
+	Callback string
+	// Content is the task content to deliver
+	Content string
+	// CreatTime is the time task created
+	CreatTime int64
+}
+
+const (
+	DelayBucket = "later_delay"
+	UnackBucket = "later_unack"
+	ErrorBucket = "later_error"
+)
+
+func createTask(task *Task) error {
+	data, err := json.Marshal(task)
+	if err != nil {
+		return err
+	}
+	c := pool.Get()
+	defer c.Close()
+	_, err = c.Do("SET", task.ID, String(data), "EX", TaskTTL)
+	if err != nil {
+		return err
+	}
+	_, err = c.Do("ZADD", DelayBucket, task.ExecuteTime, task.ID)
+	return err
+}
+
+func updateTask(task *Task) error {
+	data, err := json.Marshal(task)
+	if err != nil {
+		return err
+	}
+	c := pool.Get()
+	defer c.Close()
+	ttl, err := redis.Int(c.Do("TTL", task.ID))
+	if err != nil {
+		return err
+	}
+	_, err = c.Do("SET", task.ID, String(data), "EX", ttl)
+	return err
+}
+
+func getTask(id string) (*Task, error) {
+	c := pool.Get()
+	defer c.Close()
+	data, err := redis.String(c.Do("GET", id))
+	if err != nil {
+		return nil, err
+	}
+	var task Task
+	err = json.Unmarshal(Slice(data), &task)
+	return &task, err
+}
+
+func getTasks(bucket string, begin int64, end int64) ([]string, error) {
+	c := pool.Get()
+	defer c.Close()
+	return redis.Strings(c.Do("ZRANGEBYSCORE", bucket, begin, end, "LIMIT", "0", fmt.Sprintf("%v", ZrangeOffset)))
+}
+
+func delayToUnack(id string, score int64) (bool, error) {
+	return bucketTransfer(DelayBucket, UnackBucket, id, score)
+}
+
+func unackToDelay(id string, score int64) (bool, error) {
+	return bucketTransfer(UnackBucket, DelayBucket, id, score)
+}
+
+func errorToDelay(id string, score int64) (bool, error) {
+	return bucketTransfer(ErrorBucket, DelayBucket, id, score)
+}
+
+func bucketTransfer(from string, to string, id string, score int64) (bool, error) {
+	c := pool.Get()
+	defer c.Close()
+	reply, err := redis.Int(c.Do("ZADD", to, score, id))
+	if err != nil {
+		return false, err
+	}
+	if reply == 0 {
+		return false, nil
+	}
+	_, err = c.Do("ZREM", from, id)
+	return true, err
+}
+
+func unackToError(id string, score int64) error {
+	c := pool.Get()
+	defer c.Close()
+	_, err := c.Do("ZADD", ErrorBucket, score, id)
+	if err != nil {
+		return err
+	}
+	_, err = c.Do("ZREM", UnackBucket, id)
+	return err
+}
+
+func deleteTask(id string) error {
+	c := pool.Get()
+	defer c.Close()
+	_, err := c.Do("DEL", id)
+	if err != nil {
+		return err
+	}
+	_, err = c.Do("ZREM", DelayBucket, id)
+	if err != nil {
+		return err
+	}
+	_, err = c.Do("ZREM", UnackBucket, id)
+	if err != nil {
+		return err
+	}
+	_, err = c.Do("ZREM", ErrorBucket, id)
+	return err
+}