新增：rpc 无服务注册处理

git-svn-id: svn://47.119.165.148/zhub@130 e63fbceb-bcc3-4977-ac22-735b83d8d0f4

zsub/msg-consumer.go

@@ -1,6 +1,7 @@
 package zsub
 
 import (
+	"encoding/json"
 	"log"
 	"strconv"
 	"strings"
@@ -61,6 +62,27 @@ func msgAccept(v Message) {
 	case "groupid":
 		c.groupid = rcmd[1]
 		return
+	case "rpc":
+		// if rpc and  no sub back error
+		if zsub.noSubscribe(rcmd[1]) {
+			rpcBody := make(map[string]string)
+			json.Unmarshal([]byte(rcmd[2]), &rpcBody)
+			log.Println("rpc no subscribe: ", rcmd[1])
+
+			ruk := rpcBody["ruk"]
+			zsub.Publish(strings.Split(ruk, "::")[0], "{'retcode': 404, 'retinfo': '服务离线！', 'ruk': '"+ruk+"'}")
+			return
+		}
+
+		if len(rcmd) != 3 {
+			c.send("-Error: publish para number![" + strings.Join(rcmd, " ") + "]")
+		} else {
+			if len(topicChan) < cap(topicChan) {
+				topicChan <- rcmd
+			}
+			zsub.Publish(rcmd[1], rcmd[2])
+		}
+		return
 	case "publish":
 		if len(rcmd) != 3 {
 			c.send("-Error: publish para number![" + strings.Join(rcmd, " ") + "]")

zsub/zsub.go

@@ -510,6 +510,20 @@ func (s *ZSub) Clearup() {
 	}
 }
 
+func (s *ZSub) noSubscribe(topic string) bool {
+	zTopic := s.topics[topic]
+	if zTopic == nil || len(zTopic.groups) == 0 {
+		return true
+	}
+
+	for _, g := range zTopic.groups {
+		if len(g.conns) > 0 {
+			return false
+		}
+	}
+	return true
+}
+
 func ZSubx() *ZSub {
 	return zsub
 }