mirror/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-06-12 20:59:54 +08:00
Code Issues Packages Projects Releases Wiki Activity

use sessionId

Browse Source
This commit is contained in:
chrislu 2025-02-25 01:15:30 -08:00
parent 2ebb9927ae
commit 0129a7bf9b
1 changed files with 8 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
weed/mq/agent/agent_grpc_subscribe.go
View File

@ -14,15 +14,16 @@ import (
func (a *MessageQueueAgent) SubscribeRecord(stream mq_agent_pb.SeaweedMessagingAgent_SubscribeRecordServer) error { func (a *MessageQueueAgent) SubscribeRecord(stream mq_agent_pb.SeaweedMessagingAgent_SubscribeRecordServer) error {
// the first message is the subscribe request // the first message is the subscribe request
// it should only contain the session id // it should only contain the session id
m, err := stream.Recv() initMessage, err := stream.Recv()
if err != nil { if err != nil {
return err return err
} }
sessionId := SessionId(initMessage.SessionId)
a.subscribersLock.RLock() a.subscribersLock.RLock()
subscriberEntry, found := a.subscribers[SessionId(m.SessionId)] subscriberEntry, found := a.subscribers[sessionId]
a.subscribersLock.RUnlock() a.subscribersLock.RUnlock()
if !found { if !found {
return fmt.Errorf("subscribe session id %d not found", m.SessionId) return fmt.Errorf("subscribe session id %d not found", sessionId)
} }
defer func() { defer func() {
subscriberEntry.lastActiveTsNs = time.Now().UnixNano() subscriberEntry.lastActiveTsNs = time.Now().UnixNano()
@ -34,6 +35,7 @@ func (a *MessageQueueAgent) SubscribeRecord(stream mq_agent_pb.SeaweedMessagingA
record := &schema_pb.RecordValue{} record := &schema_pb.RecordValue{}
err := proto.Unmarshal(m.Data.Value, record) err := proto.Unmarshal(m.Data.Value, record)
if err != nil { if err != nil {
glog.V(0).Infof("unmarshal record value: %v", err)
if lastErr == nil { if lastErr == nil {
lastErr = err lastErr = err
} }
@ -44,6 +46,7 @@ func (a *MessageQueueAgent) SubscribeRecord(stream mq_agent_pb.SeaweedMessagingA
Value: record, Value: record,
TsNs: m.Data.TsNs, TsNs: m.Data.TsNs,
}); sendErr != nil { }); sendErr != nil {
glog.V(0).Infof("send record: %v", sendErr)
if lastErr == nil { if lastErr == nil {
lastErr = sendErr lastErr = sendErr
} }
@ -53,7 +56,7 @@ func (a *MessageQueueAgent) SubscribeRecord(stream mq_agent_pb.SeaweedMessagingA
go func() { go func() {
subErr := subscriberEntry.entry.Subscribe() subErr := subscriberEntry.entry.Subscribe()
if subErr != nil { if subErr != nil {
glog.V(0).Infof("subscriber %d subscribe: %v", m.SessionId, subErr) glog.V(0).Infof("subscriber %d subscribe: %v", sessionId, subErr)
if lastErr == nil { if lastErr == nil {
lastErr = subErr lastErr = subErr
} }
@ -63,6 +66,7 @@ func (a *MessageQueueAgent) SubscribeRecord(stream mq_agent_pb.SeaweedMessagingA
for { for {
m, err := stream.Recv() m, err := stream.Recv()
if err != nil { if err != nil {
glog.V(0).Infof("subscriber %d receive: %v", sessionId, err)
return err return err
} }
if m != nil { if m != nil {