From 81afb6ffbf7f76f49644a1832dcfe241552d7e08 Mon Sep 17 00:00:00 2001
From: liuxiaolong <liuxiaolong@aiotlink.com>
Date: 星期五, 05 二月 2021 18:20:09 +0800
Subject: [PATCH] add recvandsend
---
micronode.go | 83 ++++++++++++++++++++++++++---------------
1 files changed, 52 insertions(+), 31 deletions(-)
diff --git a/micronode.go b/micronode.go
index 3d83727..9fa49a2 100644
--- a/micronode.go
+++ b/micronode.go
@@ -90,21 +90,24 @@
ms.handlers = funcMap
go ms.startHeartbeat()
+ //鎺ユ敹璁㈤槄鍒扮殑娑堟伅
+ go ms.startRecvSubMsg()
+ //浣滀负server鍚姩
+ ms.serve()
+}
+//寮�濮嬫帴鏀惰闃呮秷鎭�
+func (ms *MicroNode) startRecvSubMsg() {
for {
select {
case <- ms.ctx.Done():
return
default:
- msgS, msgR, keyR := ms.handle.GetMsg()
+ msgS := ms.handle.GetMsg()
if msgS != nil {
//鏀跺埌鍏跺畠杩涚▼鐨勫彂甯冩秷鎭�
ms.printLog("Recv Sub Message:", string(msgS.Body))
ms.SubCh <- msgS
- }
- if msgR != nil {
- //鏀跺埌鍏跺畠杩涚▼鐨勮姹傛秷鎭�
- go ms.serve(msgR, keyR)
}
time.Sleep(50 * time.Millisecond)
@@ -189,39 +192,49 @@
return nil, fmt.Errorf("GetRegisteredClient list failed")
}
-func (ms *MicroNode) serve(msgR *MsgInfo, p int) {
+func (ms *MicroNode) call(rdata []byte, rkey int, sdata *[]byte) bool {
+ ri := &Reply{}
if ms.handlers == nil {
- return
- }
-
- var reqBody Request
- var ri *Reply
- err := json.Unmarshal(msgR.Body, &reqBody)
- if err != nil {
- ms.printLog("serve unmarshal msgR.Body err:", err)
- ri = &Reply {
- Msg: err.Error(),
- }
+ ri.Msg = "send wrong addr, check yourself!!!"
} else {
- ms.printLog("reqBody Path:", reqBody.Path, " contentType:", reqBody.ContentType, " formMap:",reqBody.FormMap, " postFormMap:", reqBody.PostFormMap, "to key: ", p)
-
- if f,ok := ms.handlers[reqBody.Path];ok {
- reqBody.SrcProc = msgR.SrcProc
- ri = f(&reqBody)
- ms.printLog("call funcMap f,reply:", *ri)
+ var msgR MsgInfo
+ err := json.Unmarshal(rdata, &msgR)
+ if err != nil {
+ ri.Msg = err.Error()
} else {
- ms.printLog("ms.funcMap not eixst path: ", reqBody.Path)
- ri = &Reply{
- Success: false,
- Msg: "璇锋眰鐨勬帴鍙ｄ笉瀛樺湪锛岃妫�鏌rl",
- Data: "璇锋眰鐨勬帴鍙ｄ笉瀛樺湪锛岃妫�鏌rl",
+ var reqBody Request
+ err = json.Unmarshal(rdata, &msgR.Body)
+ if err != nil {
+ ri.Msg = err.Error()
+ } else {
+ ms.printLog("srcProc:", reqBody.SrcProc,"reqBody Path:", reqBody.Path, " contentType:", reqBody.ContentType, " formMap:",reqBody.FormMap, " postFormMap:", reqBody.PostFormMap, "to key: ", rkey)
+ if f,ok := ms.handlers[reqBody.Path];ok {
+ reqBody.SrcProc = msgR.SrcProc
+ ri = f(&reqBody)
+ ms.printLog("call funcMap f,reply:", *ri)
+ } else {
+ ms.printLog("ms.funcMap not eixst path: ", reqBody.Path)
+ ri.Msg = "璇锋眰鐨勬帴鍙ｄ笉瀛樺湪锛岃妫�鏌rl"
+ }
}
}
}
+ result, err := json.Marshal(*ri)
+ if err != nil {
+ sdata = nil
+ } else {
+ sdata = &result
+ }
+ return ri.Success
+}
- retErr := ms.handle.Reply(p, ri)
- if retErr != nil {
- ms.printLog("retErr:", retErr)
+func (ms *MicroNode) serve() {
+ if ms.handlers == nil {
+ return
+ }
+ for i:=0;i<10;i++ {
+ ms.handle.wg.Add(1)
+ go recvandsendRoutine(ms.ctx, ms.handle.sockRep.sock, ms.handle.wg, ms.call, ms.fnLog)
}
}
@@ -241,6 +254,14 @@
return ms.handle.Pub(nodes, pi)
}
+func (ms *MicroNode) PublishNetTimeout(nodes []bhomebus.NetNode, topic string, msg []byte, timeout int) int {
+ pi := &MsgInfo{
+ Topic: topic,
+ Body: msg,
+ }
+ return ms.handle.PubTimeout(nodes, pi, timeout)
+}
+
//璁㈤槄涓婚
func (ms *MicroNode) Subscribe(topics []string) {
ms.handle.Sub(topics)
--
Gitblit v1.8.0