init

src/mq/handle/callback_queue.go

@@ -0,0 +1,67 @@
+package handle
+
+import (
+	"context"
+	"errors"
+	"github.com/assimon/luuu/config"
+	"github.com/assimon/luuu/model/data"
+	"github.com/assimon/luuu/model/mdb"
+	"github.com/assimon/luuu/model/response"
+	"github.com/assimon/luuu/util/http_client"
+	"github.com/assimon/luuu/util/json"
+	"github.com/assimon/luuu/util/log"
+	"github.com/assimon/luuu/util/sign"
+	"github.com/hibiken/asynq"
+)
+
+const QueueOrderCallback = "order:callback"
+
+func NewOrderCallbackQueue(order *mdb.Orders) (*asynq.Task, error) {
+	payload, err := json.Cjson.Marshal(order)
+	if err != nil {
+		return nil, err
+	}
+	return asynq.NewTask(QueueOrderCallback, payload), nil
+}
+
+func OrderCallbackHandle(ctx context.Context, t *asynq.Task) error {
+	var order mdb.Orders
+	err := json.Cjson.Unmarshal(t.Payload(), &order)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		if err := recover(); err != nil {
+			log.Sugar.Error(err)
+		}
+	}()
+	defer func() {
+		data.SaveCallBackOrdersResp(&order)
+	}()
+	client := http_client.GetHttpClient()
+	orderResp := response.OrderNotifyResponse{
+		TradeId:            order.TradeId,
+		OrderId:            order.OrderId,
+		Amount:             order.Amount,
+		ActualAmount:       order.ActualAmount,
+		Token:              order.Token,
+		BlockTransactionId: order.BlockTransactionId,
+		Status:             mdb.StatusPaySuccess,
+	}
+	signature, err := sign.Get(orderResp, config.GetApiAuthToken())
+	if err != nil {
+		return err
+	}
+	orderResp.Signature = signature
+	resp, err := client.R().SetHeader("powered-by", "Epusdt(https://github.com/assimon/epusdt)").SetBody(orderResp).Post(order.NotifyUrl)
+	if err != nil {
+		return err
+	}
+	body := string(resp.Body())
+	if body != "ok" {
+		order.CallBackConfirm = mdb.CallBackConfirmNo
+		return errors.New("not ok")
+	}
+	order.CallBackConfirm = mdb.CallBackConfirmOk
+	return nil
+}

src/mq/handle/order_expiration_queue.go

@@ -0,0 +1,37 @@
+package handle
+
+import (
+	"context"
+	"fmt"
+	"github.com/assimon/luuu/model/data"
+	"github.com/assimon/luuu/model/mdb"
+	"github.com/hibiken/asynq"
+)
+
+const QueueOrderExpiration = "order:expiration"
+
+func NewOrderExpirationQueue(tradeId string) (*asynq.Task, error) {
+	return asynq.NewTask(QueueOrderExpiration, []byte(tradeId)), nil
+}
+
+// OrderExpirationHandle 设置订单过期
+func OrderExpirationHandle(ctx context.Context, t *asynq.Task) error {
+	tradeId := string(t.Payload())
+	orderInfo, err := data.GetOrderInfoByTradeId(tradeId)
+	if err != nil {
+		return err
+	}
+	if orderInfo.ID <= 0 || orderInfo.Status != mdb.StatusWaitPay {
+		return nil
+	}
+	err = data.UpdateOrderIsExpirationById(orderInfo.ID)
+	if err != nil {
+		return err
+	}
+	actualAmount := fmt.Sprintf("%.4f", orderInfo.ActualAmount)
+	err = data.ClearPayCache(orderInfo.Token, actualAmount)
+	if err != nil {
+		return err
+	}
+	return nil
+}

src/mq/queue.go

@@ -0,0 +1,51 @@
+package mq
+
+import (
+	"fmt"
+	"github.com/assimon/luuu/mq/handle"
+	"github.com/assimon/luuu/util/log"
+	"github.com/hibiken/asynq"
+	"github.com/spf13/viper"
+)
+
+var MClient *asynq.Client
+
+func Start() {
+	redis := asynq.RedisClientOpt{
+		Addr: fmt.Sprintf(
+			"%s:%s",
+			viper.GetString("redis_host"),
+			viper.GetString("redis_port")),
+		DB:       viper.GetInt("redis_db"),
+		Password: viper.GetString("redis_passwd"),
+	}
+	initClient(redis)
+	go initListen(redis)
+}
+
+func initClient(redis asynq.RedisClientOpt) {
+	MClient = asynq.NewClient(redis)
+}
+
+func initListen(redis asynq.RedisClientOpt) {
+	srv := asynq.NewServer(
+		redis,
+		asynq.Config{
+			// Specify how many concurrent workers to use
+			Concurrency: viper.GetInt("queue_concurrency"),
+			// Optionally specify multiple queues with different priority.
+			Queues: map[string]int{
+				"critical": viper.GetInt("queue_level_critical"),
+				"default":  viper.GetInt("queue_level_default"),
+				"low":      viper.GetInt("queue_level_low"),
+			},
+			Logger: log.Sugar,
+		},
+	)
+	mux := asynq.NewServeMux()
+	mux.HandleFunc(handle.QueueOrderExpiration, handle.OrderExpirationHandle)
+	mux.HandleFunc(handle.QueueOrderCallback, handle.OrderCallbackHandle)
+	if err := srv.Run(mux); err != nil {
+		log.Sugar.Fatalf("[queue] could not run server: %v", err)
+	}
+}