添加DeSub，优化订阅主题消息通道

liuxiaolong

2020-12-30 b93b67a7e237cbf59569dcaacfa84257856ba16e

 micronode.go

@@ -19,7 +19,7 @@
   serverId    string
   fnLog       func(...interface{})

   SubChM       map[string]chan *MsgInfo //以订阅的主题为key
   SubCh       chan *MsgInfo
}

func NewMicroNode(ctx context.Context,q chan os.Signal, serverId string, reg *RegisterInfo, procInfo *ProcInfo,fnLog func(v ...interface{})) (*MicroNode, error){
@@ -33,11 +33,8 @@
      handle:   handle,
      reg:      reg,
      procInfo: procInfo,
      fnLog: fnLog,
      SubChM:   make(map[string]chan *MsgInfo),
   }
   for _,subTopic := range reg.SubTopic {
      mn.SubChM[subTopic] = make(chan *MsgInfo, 512)
      fnLog:    fnLog,
      SubCh:    make(chan *MsgInfo, 512),
   }

   return mn, nil
@@ -102,9 +99,7 @@
         if msgS != nil {
            //收到其它进程的发布消息
            ms.printLog("Recv Sub Message:", string(msgS.Body))
            if ch,ok := ms.SubChM[msgS.Topic];ok {
               ch <- msgS
            }
            ms.SubCh <- msgS
         }
         if msgR != nil {
            //收到其它进程的请求消息
@@ -259,8 +254,46 @@
   return ms.handle.Pub(nodes, pi)
}

//订阅主题
func (ms *MicroNode) Subscribe(topics []string) {
   ms.handle.Sub(topics)
   for _,t := range topics {
      if ms.reg.SubTopic == nil {
         ms.reg.SubTopic = make([]string, 0)
      }
      found := false
      for _,it := range ms.reg.SubTopic {
         if it == t {
            found = true
            break
         }
      }
      if !found {
         ms.reg.SubTopic = append(ms.reg.SubTopic, t)
      }
   }
}

//取消订阅的主题
func (ms *MicroNode) DeSub(topics []string) {
   ms.printLog("DeSub topics:", topics)
   ms.handle.DeSub(topics)
   if ms.reg.SubTopic != nil {
      var leftTopics []string
      for _,t := range ms.reg.SubTopic {
         found := false
         for _,it := range topics {
            if it == t {
               found = true
               break
            }
         }
         if !found {
            leftTopics = append(leftTopics, t)
         }
      }
      ms.reg.SubTopic = leftTopics
   }
}

//free handle