diff options
| author | Chris Lu <chris.lu@gmail.com> | 2020-05-16 08:57:29 -0700 |
|---|---|---|
| committer | Chris Lu <chris.lu@gmail.com> | 2020-05-16 08:57:29 -0700 |
| commit | b0de01ff3b91cf28edbbaa536eed0e02db4d492c (patch) | |
| tree | a659404528abd4164b051049c43d4de83242ee11 /weed/messaging/msgclient/sub_chan.go | |
| parent | 7aafb9e3f800a2cf2ad704b291887b9ea89507d4 (diff) | |
| download | seaweedfs-b0de01ff3b91cf28edbbaa536eed0e02db4d492c.tar.xz seaweedfs-b0de01ff3b91cf28edbbaa536eed0e02db4d492c.zip | |
able to detect disconnected subscribers
Diffstat (limited to 'weed/messaging/msgclient/sub_chan.go')
| -rw-r--r-- | weed/messaging/msgclient/sub_chan.go | 6 |
1 files changed, 5 insertions, 1 deletions
diff --git a/weed/messaging/msgclient/sub_chan.go b/weed/messaging/msgclient/sub_chan.go index 5465c5913..3eabc6210 100644 --- a/weed/messaging/msgclient/sub_chan.go +++ b/weed/messaging/msgclient/sub_chan.go @@ -27,7 +27,7 @@ func (mc *MessagingClient) NewSubChannel(subscriberId, chanName string) (*SubCha if err != nil { return nil, err } - sc, err := setupSubscriberClient(grpcConnection, subscriberId, "chan", chanName, 0, time.Unix(0, 0)) + sc, err := setupSubscriberClient(grpcConnection, tp, subscriberId, time.Unix(0, 0)) if err != nil { return nil, err } @@ -48,6 +48,10 @@ func (mc *MessagingClient) NewSubChannel(subscriberId, chanName string) (*SubCha log.Printf("fail to receive from netchan %s: %v", chanName, subErr) return } + if resp.Data == nil { + // this could be heartbeat from broker + continue + } if resp.Data.IsClose { t.stream.Send(&messaging_pb.SubscriberMessage{ IsClose: true, |