初步完成数据交流.

proxyserver/.gitignore

@@ -0,0 +1 @@
+proxyserver

proxyserver/go.mod

@@ -3,9 +3,12 @@ module git.nonolive.co/eson.hsm/proxyserver
 go 1.15
 
 require (
+	github.com/474420502/focus v0.12.0
+	github.com/bwmarrin/snowflake v0.3.0
 	github.com/gin-gonic/gin v1.6.3
 	github.com/go-playground/validator/v10 v10.4.1 // indirect
 	github.com/golang/protobuf v1.4.3 // indirect
+	github.com/google/uuid v1.1.2
 	github.com/json-iterator/go v1.1.10 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.1 // indirect

proxyserver/go.sum

@@ -1,8 +1,14 @@
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
+github.com/474420502/focus v0.12.0 h1:+icbmj7IEOefvTegHt5EpcHt6WFbe2miIrceUJx2Evo=
+github.com/474420502/focus v0.12.0/go.mod h1:d0PMjtMxFz1a9HIhwyFPkWa+JF+0LgOrEUfd8iZka6s=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
+github.com/Pallinder/go-randomdata v1.1.0/go.mod h1:yHmJgulpD2Nfrm0cR9tI/+oAgRqCQQixsA8HyRZfV9Y=
+github.com/bwmarrin/snowflake v0.3.0 h1:xm67bEhkKh6ij1790JB83OujPR5CzNe8QuQqAgISZN0=
+github.com/bwmarrin/snowflake v0.3.0/go.mod h1:NdZxfVWX+oR6y2K0o6qAYv6gIOP9rjG0/E9WsDpxqwE=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
@@ -39,6 +45,8 @@ github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMyw
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
+github.com/google/uuid v1.1.2 h1:EVhdT+1Kseyi1/pUmXKaFxYsDNy9RQYkMWRH68J/W7Y=
+github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/json-iterator/go v1.1.9 h1:9yzud/Ht36ygwatGx56VwCZtlI/2AD15T1X2sjSuGns=
 github.com/json-iterator/go v1.1.9/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
 github.com/json-iterator/go v1.1.10 h1:Kz6Cvnvv2wGdaG/V8yMvfkmNiXq9Ya2KUv4rouJJr68=

proxyserver/main_test.go

@@ -0,0 +1,13 @@
+package main
+
+import (
+	"testing"
+)
+
+func TestMain(t *testing.T) {
+	main()
+}
+
+func TestTask(t *testing.T) {
+
+}

proxyserver/oplog.go

@@ -0,0 +1,38 @@
+package main
+
+import (
+	"encoding/json"
+	"log"
+	"os"
+)
+
+var oplog *Oplog
+
+// Oplog 操作日志类
+type Oplog struct {
+	oplog *log.Logger
+}
+
+func initOplog() {
+	if oplog == nil {
+		oplog = &Oplog{}
+		f, err := os.OpenFile("./op.log", os.O_CREATE|os.O_APPEND|os.O_RDWR, 0644)
+		if err != nil {
+			panic(err)
+		}
+		oplog.oplog = log.New(f, "", log.Lmsgprefix)
+	}
+}
+
+// Write 操作日志写入
+func (op *Oplog) Write(data interface{}) {
+	if d, ok := data.(*Task); ok {
+		data, err := json.Marshal(d)
+		if err != nil {
+			log.Println(err)
+		}
+		op.oplog.Println(string(data))
+		return
+	}
+	log.Println("data must gin.H: ", data)
+}

proxyserver/queue.go

@@ -0,0 +1,55 @@
+package main
+
+import (
+	"sync"
+
+	pqueuekey "github.com/474420502/focus/priority_queuekey"
+)
+
+// Queue 存储队列
+type Queue struct {
+	queue *pqueuekey.PriorityQueue
+	lock  sync.Mutex
+}
+
+// NewQueue 创建队列
+func NewQueue() *Queue {
+	q := &Queue{}
+	q.queue = pqueuekey.New(CompareTaskID)
+	return q
+}
+
+// Get 获取该值是否存在
+func (q *Queue) Get(key interface{}) (result interface{}, ok bool) {
+	q.lock.Lock()
+	defer q.lock.Unlock()
+	return q.queue.Get(key)
+}
+
+// Push 入队
+func (q *Queue) Push(key interface{}, value interface{}) {
+	q.lock.Lock()
+	defer q.lock.Unlock()
+	q.queue.Push(key, value)
+}
+
+// Pop 出队
+func (q *Queue) Pop() (result interface{}, ok bool) {
+	q.lock.Lock()
+	defer q.lock.Unlock()
+	return q.queue.Pop()
+}
+
+// Remove 删除指定key
+func (q *Queue) Remove(key interface{}) (result interface{}, ok bool) {
+	q.lock.Lock()
+	defer q.lock.Unlock()
+	return q.queue.Remove(key)
+}
+
+// Top 队头
+func (q *Queue) Top() (result interface{}, ok bool) {
+	q.lock.Lock()
+	defer q.lock.Unlock()
+	return q.queue.Top()
+}

proxyserver/router.go

@@ -0,0 +1,130 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"net/http"
+	"time"
+
+	"github.com/bwmarrin/snowflake"
+
+	"github.com/gin-gonic/gin"
+)
+
+var taskQueue = NewQueue()
+var waitQueue = NewQueue()
+var readyQueue = NewQueue()
+
+var snowNode *snowflake.Node
+
+func init() {
+	log.SetFlags(log.Llongfile | log.LstdFlags)
+	initOplog()
+	// Create a new Node with a Node number of 1
+	node, err := snowflake.NewNode(1)
+	if err != nil {
+		panic(err)
+	}
+	snowNode = node
+
+	engine.GET("/", func(c *gin.Context) {
+		c.JSON(200, Response{Code: 200, Message: "", Data: nil})
+	})
+
+	task := engine.Group("/task")
+	task.GET("/get", GetTask)
+	task.POST("/put", PutTask)
+
+	task.POST("/content", ContentTask)
+	task.GET("/ready", ReadyTask)
+}
+
+// GetTask 获取当前一条任务列表
+func GetTask(c *gin.Context) {
+	if result, ok := taskQueue.Pop(); ok {
+		c.JSON(http.StatusOK, Response{Code: 200, Message: "", Data: result})
+		return
+	}
+	c.JSON(http.StatusOK, Response{Code: 204, Message: "No Task"})
+}
+
+// PutTask 把一条任务放入队列
+func PutTask(c *gin.Context) {
+
+	u := c.PostForm("url")
+	if u != "" {
+		data := NewTask()
+
+		now := time.Now()
+		tid := snowNode.Generate().Base64()
+		data.Store("taskid", tid)
+		data.Store("url", u)
+		data.Store("ts", now.UnixNano())
+		taskQueue.Push(tid, data)
+		oplog.Write(data)
+		c.JSON(http.StatusOK, Response{Code: 200, Message: "ok", Data: data})
+		return
+	}
+
+	c.JSON(http.StatusOK, Response{Code: 400, Message: "url 不错在"})
+	return
+}
+
+// ContentTask 把一条任务放入队列
+func ContentTask(c *gin.Context) {
+	r := c.PostForm("response")
+	response := &Response{}
+	json.Unmarshal([]byte(r), response)
+	if response.Code == 200 {
+		data := response.Data.(gin.H)
+		tid := data["taskid"]
+		if iv, ok := waitQueue.Remove(tid); ok {
+			task := iv.(*Task)
+			task.Store("content", data["content"])
+			task.Store("is_read", "false")
+			task.Store("status", "ready")
+			readyQueue.Push(tid, task) // 进入回调发送队列.TODO: 内容持久化
+		}
+	}
+}
+
+// AckTask 确认整个任务流程完成.
+func AckTask(c *gin.Context) {
+	tid := c.Query("taskid")
+	if tid != "" {
+		if itask, ok := readyQueue.Get(tid); ok {
+			task := itask.(*Task)
+			if status, ok := task.Load("status"); ok {
+				if status.(string) == "readying" {
+					task.Store("status", "readied")
+					c.JSON(http.StatusOK, Response{Code: 200, Message: fmt.Sprintf("task %s readied", tid)})
+					return
+				}
+				c.JSON(http.StatusOK, Response{Code: 200, Message: fmt.Sprintf("task %s is not readying", tid)})
+				return
+			}
+		}
+	} else {
+		c.JSON(http.StatusOK, Response{Code: 404, Message: fmt.Sprintf("query taskid params must exist")})
+		return
+	}
+	c.JSON(http.StatusOK, Response{Code: 404, Message: fmt.Sprintf("taskid: %s is not found", tid)})
+}
+
+// ReadyTask 把一条任务放入队列
+func ReadyTask(c *gin.Context) {
+	tid := c.Query("taskid")
+	if tid != "" {
+		if itask, ok := readyQueue.Get(tid); ok {
+			task := itask.(*Task)
+			task.Store("status", "readying")
+			c.JSON(http.StatusOK, Response{Code: 200, Data: task})
+			return
+		}
+	} else {
+		c.JSON(http.StatusOK, Response{Code: 404, Message: fmt.Sprintf("query taskid params must exist")})
+		return
+	}
+	c.JSON(http.StatusOK, Response{Code: 404, Message: fmt.Sprintf("taskid: %s is not found", tid)})
+}

proxyserver/struct.go

@@ -0,0 +1,46 @@
+package main
+
+import (
+	"encoding/json"
+	"sync"
+
+	"github.com/gin-gonic/gin"
+)
+
+// Response 统一的返回格式
+type Response struct {
+	Code    int         `json:"code"`
+	Message string      `json:"msg"`
+	Data    interface{} `json:"data"`
+}
+
+// Task 任务
+type Task struct {
+	data gin.H
+	lock sync.Mutex
+}
+
+func NewTask() *Task {
+	return &Task{data: gin.H{}}
+}
+
+// Store 存储一个字段
+func (t *Task) Store(key string, value interface{}) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	t.data[key] = value
+}
+
+// Load 取一个字段
+func (t *Task) Load(key string) (value interface{}, ok bool) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	value, ok = t.data[key]
+	return
+}
+
+func (t *Task) MarshalJSON() ([]byte, error) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	return json.Marshal(t.data)
+}

proxyserver/uidcompare.go

@@ -0,0 +1,40 @@
+package main
+
+// CompareTaskID 任务id比较
+func CompareTaskID(k1, k2 interface{}) int {
+	s1 := k2.(string)
+	s2 := k1.(string)
+
+	switch {
+	case len(s1) > len(s2):
+		for i := 0; i < len(s2); i++ {
+			if s1[i] != s2[i] {
+				if s1[i] > s2[i] {
+					return 1
+				}
+				return -1
+			}
+		}
+		return 1
+	case len(s1) < len(s2):
+		for i := 0; i < len(s1); i++ {
+			if s1[i] != s2[i] {
+				if s1[i] > s2[i] {
+					return 1
+				}
+				return -1
+			}
+		}
+		return -1
+	default:
+		for i := 0; i < len(s1); i++ {
+			if s1[i] != s2[i] {
+				if s1[i] > s2[i] {
+					return 1
+				}
+				return -1
+			}
+		}
+		return 0
+	}
+}