| | |
| | | "sync" |
| | | "time" |
| | | "unsafe" |
| | | |
| | | "github.com/bytedance/sonic" |
| | | ) |
| | | |
| | | type MsgReq struct { |
| | |
| | | |
| | | handle.printLog("register done!" ) |
| | | |
| | | handle.wg = &sync.WaitGroup{} |
| | | |
| | | //有订阅消息才需要启动协程接收消息 |
| | | if len(ri.SubTopic) > 0 { |
| | | handle.printLog("sub topics") |
| | |
| | | } |
| | | } |
| | | |
| | | //if len(ri.SubTopic) > 0 || len(ri.SubNetTopic) > 0 { |
| | | if len(ri.SubTopic) > 0 || len(ri.SubNetTopic) > 0 { |
| | | //启动订阅信息接收 |
| | | handle.wg.Add(1) |
| | | go recvSubRoutine(ctx, handle.wg, handle.ChSub, handle.printLog) |
| | | //} |
| | | } |
| | | |
| | | return handle, nil |
| | | } |
| | |
| | | for { |
| | | select { |
| | | case <-ctx.Done(): |
| | | logFn("recvRoutine ctx.Done") |
| | | logFn("recvSubRoutine ctx.Done") |
| | | wg.Done() |
| | | return |
| | | default: |
| | | if bhsgo.ReadSub(&procId, &msg, -1) { |
| | | if bhsgo.ReadSub(&procId, &msg, 100) { |
| | | ch <- msg |
| | | logFn("ReadSub topic:", string(msg.Topic), " data:", string(msg.Data)) |
| | | logFn("ReadSub topic:", string(msg.Topic), " len(data):", len(msg.Data)) |
| | | |
| | | procId = "" |
| | | msg.Reset() |
| | |
| | | if bhsgo.Request(&dest, req, &pid, &mrt, milliSecs) { |
| | | var reply Reply |
| | | if err := json.Unmarshal(mrt.Data, &reply); err != nil { |
| | | h.printLog("bhsgo.Request ret true, but unmarshal err:", err, " mrt.Data:", string(mrt.Data)) |
| | | return nil,err |
| | | } |
| | | |
| | |
| | | } |
| | | |
| | | func (h *BHBus) Reply(src unsafe.Pointer, i *Reply) error { |
| | | data,err := json.Marshal(*i) |
| | | defer func() { |
| | | i = nil |
| | | }() |
| | | |
| | | data,err := sonic.Marshal(i) |
| | | if err != nil { |
| | | return err |
| | | } |
| | | rep := bhome_msg.MsgRequestTopicReply{ |
| | | Data: data, |
| | | } |
| | | |
| | | if bhsgo.SendReply(src, &rep) { |
| | | return nil |
| | | } |