增加怡亚通队列

client.go

@@ -0,0 +1,65 @@
+package rabbitmqrpc
+
+import (
+	"encoding/json"
+	"log"
+	"strconv"
+	"time"
+
+	"git.tetele.net/tgo/crypter"
+
+	"github.com/golang/protobuf/proto"
+)
+
+func Send(exchange, key string, data interface{}, url ...string) (*MsgSend, error) {
+
+	conn, err := rpc_server_conn(url...)
+	if err != nil {
+		return nil, err
+	}
+	defer conn.Close()
+
+	arg_data, err := json.Marshal(data)
+	if err != nil {
+		log.Println("json marshal error:", err)
+	}
+
+	arg := MsgBody{}
+	arg.Exchange = exchange
+	arg.Key = key
+	arg.Data = string(arg_data)
+
+	data_json, err := json.Marshal(arg)
+	if err != nil {
+		return nil, err
+	}
+	now_int64 := time.Now().Unix()
+
+	encryData := crypter.DesEn(string(data_json), DES_KEY)
+
+	now := strconv.FormatInt(now_int64, 10)
+
+	sign := Sign(encryData, now)
+
+	req := &Request{proto.String(encryData), proto.String(now), proto.String(sign), nil}
+
+	res := &Response{}
+
+	err = conn.Send(req, res)
+
+	if err != nil {
+		return nil, err
+	}
+
+	res_data, err := GetOrgData(res)
+
+	var res_arr MsgSend
+
+	err = json.Unmarshal([]byte(res_data), &res_arr)
+
+	if err != nil {
+		return nil, err
+	}
+	return &res_arr, nil
+
+}

client_test.go

@@ -0,0 +1,25 @@
+package rabbitmqrpc
+
+import (
+	"testing"
+)
+
+func Test_Send(t *testing.T) {
+
+	exchange := "send_msg"
+	key := "api"
+
+	msg := map[string]interface{}{
+		"site_id": "100002",
+		"dbname":  "testing",
+		"data": map[string]string{
+			"order_id": "111",
+			"order_sn": "cccc",
+		},
+	}
+
+	ret, err := Send(exchange, key, msg)
+
+	t.Log(ret)
+	t.Log(err)
+}

common.go

@@ -1,5 +1,9 @@
 package rabbitmqrpc
 
+import (
+	"git.tetele.net/tgo/conf"
+)
+
 const DES_KEY = "mqbitrab"
 
 type MsgBody struct {
@@ -7,14 +11,161 @@ type MsgBody struct {
 	Key      string
 	Data     string
 }
+type DelayMsgBody struct {
+	Exchange string
+	Key      string
+	Data     string
+	Delay    int64
+}
+type MsgSend struct {
+	Result string
+}
+
+//订单路由
+const ORDER_EXCHANGE string = "ex_order"
+
+//已创建订单route key
+const ORDER_CREATED_KEY string = "created"
+
+//订单已创建队列，订单服务用
+const ORDER_CREATED_QUEUE = "order_created_queue"
+
+//已创建订单队列，串货服务用
+const CHANNEL_ORDER_CREATED_QUEUE = "channel_order_created_queue"
+
+//已支付订单route key
+const ORDER_PAYED_KEY string = "payed"
+
+//订单已支付队列，订单服务用
+const ORDER_PAYED_QUEUE = "order_payed_queue"
+
+//订单已支付队列，串货服务用
+const CHANNEL_ORDER_PAYED_QUEUE = "channel_order_payed_queue"
+
+//订单已支付队列，配送服务用
+const DELIVERY_ORDER_PAYED_QUEUE = "delivery_order_payed_queue"
+
+//订单已支付队列，打印服务用
+const PRINT_ORDER_PAYED_QUEUE = "print_order_payed_queue"
+
+//订单状态更新route key
+const ORDER_STATUS_CHANGE_KEY string = "status_change"
+
+//订单状态更新队列，订单服务用
+const ORDER_STATUS_UPDATE_QUEUE = "order_status_update_queue"
+
+//订单状态更新，串货服务用
+const CHANNEL_ORDER_STATUS_CHANGE_QUEUE = "channel_order_update_status"
+
+//订单取消route key
+const ORDER_CANCELED_KEY string = "canceled"
+
+//订单取消队列，订单服务用
+const ORDER_CANCELED_QUEUE = "order_canceled_queue"
+
+//订单取消队列，串货服务用
+const CHANNEL_ORDER_CANCELED_QUEUE = "channel_order_canceled_queue"
+
+//订单发货route key
+const ORDER_DELIVERED_KEY string = "delivered"
+
+//订单发货队列，订单服务用
+const ORDER_DELIVERED_QUEUE = "order_delivered_queue"
+
+//订单发货队列，串货服务用
+const CHANNEL_ORDER_DELIVERED_QUEUE = "channel_order_delivered_queue"
+
+//订单收货route key
+const ORDER_RECEIVED_KEY string = "received"
+
+//订单取消队列，订单服务用
+const ORDER_RECEIVED_QUEUE = "order_received_queue"
+
+//订单取消队列，串货服务用
+const CHANNEL_ORDER_RECEIVED_QUEUE = "channel_order_received_queue"
+
+//订单退款route key
+const ORDER_REFUNDED_KEY string = "refunded"
+
+//订单退款队列，订单服务用
+const ORDER_REFUNDED_QUEUE = "order_refunded_queue"
+
+//订单退款队列，串货服务用
+const CHANNEL_ORDER_REFUNDED_QUEUE = "channel_order_refunded_queue"
+
+//订单完成route key
+const ORDER_FINISHED_KEY string = "finished"
+
+//订单退款队列，订单服务用
+const ORDER_FINISHED_QUEUE = "order_finished_queue"
+
+//订单退款队列，串货服务用
+const CHANNEL_ORDER_FINISHED_QUEUE = "channel_order_finished_queue"
+
+//订单等待创建key，订单服务用
+const ORDER_ADD_KEY = "add"
+
+//订单等待创建队列，订单服务用
+const ORDER_ADD_QUEUE = "order_add_queue"
+
+//订单核销key
+const ORDER_VERIFY_KEY = "verify"
+
+//订单核销打印队列，打印服务用
+const PRINT_ORDER_VERIFY_QUEUE = "print_order_verify_queue"
+
+//订单延迟队列路由
+const ORDER_DELAY_EXCHANGE string = "ex_order_delay"
+
+//未支付订单过期自动取消key
+const ORDER_AUTO_CANCEL_KEY string = "order_auto_cancel"
+
+//未支付订单过期自动取消队列
+const ORDER_AUTO_CANCEL_QUEUE string = "order_auto_cancel_queue"
+
+//未支付订单过期自动取消队列，串货服务用
+const CHANNEL_ORDER_AUTO_CANCEL_QUEUE string = "channel_order_auto_cancel_queue"
+
+//订单发货后到期自动收货key
+const ORDER_AUTO_RECEIVE_KEY string = "order_auto_received"
+
+//订单发货后到期自动收货队列
+const ORDER_AUTO_RECEIVE_QUEUE string = "order_auto_receive_queue"
+
+//订单自动收货队列，串货服务用
+const CHANNEL_ORDER_AUTO_RECEIVE_QUEUE string = "channel_order_auto_receive_queue"
+
+//订单申请退款消息key
+const ORDER_ASK_FOR_REFUND_KEY string = "order_ask_for_refund"
+
+//订单申请退款消息队列
+const ORDER_ASK_FOR_REFUND_QUEUE string = "order_ask_for_refund_queue"
+
+//订单申请退款队列，串货服务用
+const CHANNEL_ORDER_ASK_FOR_REFUND_QUEUE string = "channel_order_ask_for_refund_queue"
+
+// --串货--
+//串货订单路由
+const CHANNEL_ORDER_EXCHANGE string = "ex_channel_order"
+
+//串货订单请求支付route key
+const CHANNEL_ORDER_ASKPAY_KEY = "channel_order_askpay"
+
+//订单请求支付队列，串货服务用
+const CHANNEL_ORDER_ASKPAY_QUEUE = "channel_order_askpay_queue"
 
 func rpc_server_conn(url ...string) (*RabbitmqServiceClient, error) {
 
-	var wx_rpc_url string = "127.0.0.1:7955"
+	var rpc_url string
 	if len(url) > 0 && url[0] != "" {
-		wx_rpc_url = url[0]
+		rpc_url = url[0]
+	} else if conf.RABBITMQ_RPC_URL != "" {
+		rpc_url = conf.RABBITMQ_RPC_URL
+	} else {
+		rpc_url = "127.0.0.1:" + conf.RABBITMQ_RPC_PORT
 	}
-	conn, _, err := DialRabbitmqService("tcp", wx_rpc_url)
+
+	conn, _, err := DialRabbitmqService("tcp", rpc_url)
 	if err != nil {
 		return nil, err
 	}

const_v2.go

@@ -0,0 +1,21 @@
+package rabbitmqrpc
+
+//v2
+
+//订单路由
+const ORDER_EXCHANGE_V2 string = "ex_order_v2"
+
+//已创建订单route key
+const ORDER_CREATED_KEY_V2 string = "created_v2"
+
+//订单已创建队列，订单服务用
+const ORDER_CREATED_QUEUE_V2 = "order_created_queue_v2"
+
+//已支付订单route key
+const ORDER_PAYED_KEY_V2 string = "payed_v2"
+
+//订单已支付队列，订单服务用
+const ORDER_PAYED_QUEUE_V2 = "order_payed_queue_v2"
+
+//怡亚通订单已支付队列，怡亚通订单服务用
+const ORDER_PAYED_QUEUE_V2_380STAR = "order_payed_queue_v2_380star"

go.mod

@@ -0,0 +1,11 @@
+module git.tetele.net/tgo/rabbitmqrpc
+
+go 1.14
+
+require (
+	git.tetele.net/tgo/conf v0.38.0 // indirect
+	git.tetele.net/tgo/crypter v0.2.2 // indirect
+	github.com/chai2010/protorpc v1.0.0 // indirect
+	github.com/golang/protobuf v1.5.2 // indirect
+	golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e // indirect
+)

go.sum

@@ -0,0 +1,26 @@
+git.tetele.net/tgo/conf v0.38.0 h1:mqiljzhjY9SbTmS2yL8p5clrTkkiD5JzeLrG1FT+gHM=
+git.tetele.net/tgo/conf v0.38.0/go.mod h1:AWVIBEDE5dtotthUgR0SWaR2Qa6/f+O5WQ3s7Tj8q7A=
+git.tetele.net/tgo/crypter v0.2.2 h1:YMQJh2Gj5Po4ZfelJUmXBKi01UbmtiSy3bmqRfnYQMo=
+git.tetele.net/tgo/crypter v0.2.2/go.mod h1:vfvRLZA8+lHNgNXneOcgvVhDyuv25ZRb+C6xHOmXNx0=
+github.com/chai2010/protorpc v1.0.0 h1:aJ45G9sl1utSKo35EqnBSTs5jqTpdJDJAuZMMYPAtFo=
+github.com/chai2010/protorpc v1.0.0/go.mod h1:woR3WwjaQDqFjlzdVsFEKiK5Ur12QL8mYxVPjfr5z54=
+github.com/golang/protobuf v1.0.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
+github.com/golang/protobuf v1.5.2 h1:ROPKBNFfQgOUMifHyP+KYbvpjbdoFNs+aK7DXlji0Tw=
+github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
+github.com/golang/snappy v0.0.0-20170215233205-553a64147049 h1:K9KHZbXKpGydfDN0aZrsoHpLJlZsBrGMFWbgLDGnPZk=
+github.com/golang/snappy v0.0.0-20170215233205-553a64147049/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
+github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e h1:gsTQYXdTw2Gq7RBsWvlQ91b+aEQ6bXFUngBGuR8sPpI=
+golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
+golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
+google.golang.org/protobuf v1.26.0 h1:bxAC2xTBsZGibn2RTntX0oH50xLsqy1OxA9tTL3p/lk=
+google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=

rabbitmq.pb.go

@@ -99,6 +99,7 @@ func init() {
 
 type RabbitmqService interface {
 	Send(in *Request, out *Response) error
+	SendDelay(in *Request, out *Response) error
 }
 
 // AcceptRabbitmqServiceClient accepts connections on the listener and serves requests
@@ -173,6 +174,9 @@ func NewRabbitmqServiceClient(conn io.ReadWriteCloser) (*RabbitmqServiceClient,
 func (c *RabbitmqServiceClient) Send(in *Request, out *Response) error {
 	return c.Call("RabbitmqService.Send", in, out)
 }
+func (c *RabbitmqServiceClient) SendDelay(in *Request, out *Response) error {
+	return c.Call("RabbitmqService.SendDelay", in, out)
+}
 
 // DialRabbitmqService connects to an RabbitmqService at the specified network address.
 func DialRabbitmqService(network, addr string) (*RabbitmqServiceClient, *rpc.Client, error) {

rabbitmq.proto

@@ -19,4 +19,5 @@ message Response {
 // rpc方法
 service RabbitmqService {
     rpc send (Request) returns (Response);    // 发送消息
+    rpc sendDelay (Request) returns (Response);    // 发送延迟消息
 }

send_delay.go

@@ -0,0 +1,66 @@
+package rabbitmqrpc
+
+import (
+	"encoding/json"
+	"log"
+	"strconv"
+	"time"
+
+	"git.tetele.net/tgo/crypter"
+
+	"github.com/golang/protobuf/proto"
+)
+
+func SendDelay(exchange, key string, data interface{}, delay int64, url ...string) (*MsgSend, error) {
+
+	conn, err := rpc_server_conn(url...)
+	if err != nil {
+		return nil, err
+	}
+	defer conn.Close()
+
+	arg_data, err := json.Marshal(data)
+	if err != nil {
+		log.Println("json marshal error:", err)
+	}
+
+	arg := DelayMsgBody{}
+	arg.Exchange = exchange
+	arg.Key = key
+	arg.Data = string(arg_data)
+	arg.Delay = delay
+
+	data_json, err := json.Marshal(arg)
+	if err != nil {
+		return nil, err
+	}
+	now_int64 := time.Now().Unix()
+
+	encryData := crypter.DesEn(string(data_json), DES_KEY)
+
+	now := strconv.FormatInt(now_int64, 10)
+
+	sign := Sign(encryData, now)
+
+	req := &Request{proto.String(encryData), proto.String(now), proto.String(sign), nil}
+
+	res := &Response{}
+
+	err = conn.SendDelay(req, res)
+
+	if err != nil {
+		return nil, err
+	}
+
+	res_data, err := GetOrgData(res)
+
+	var res_arr MsgSend
+
+	err = json.Unmarshal([]byte(res_data), &res_arr)
+
+	if err != nil {
+		return nil, err
+	}
+	return &res_arr, nil
+
+}

send_delay_test.go

@@ -0,0 +1,25 @@
+package rabbitmqrpc
+
+import (
+	"testing"
+)
+
+func Test_SendDelay(t *testing.T) {
+
+	exchange := "ex_order_delay"
+	key := "order_expire"
+
+	msg := map[string]interface{}{
+		"site_id": "100002",
+		"dbname":  "testing",
+		"data": map[string]string{
+			"order_id": "111",
+			"order_sn": "cccc",
+		},
+	}
+
+	ret, err := SendDelay(exchange, key, msg, 20)
+
+	t.Log(ret)
+	t.Log(err)
+}