diff options
Diffstat (limited to 'weed/server/filer_grpc_server_sub_meta.go')
| -rw-r--r-- | weed/server/filer_grpc_server_sub_meta.go | 25 |
1 files changed, 19 insertions, 6 deletions
diff --git a/weed/server/filer_grpc_server_sub_meta.go b/weed/server/filer_grpc_server_sub_meta.go index a0c53cae0..db395e3c3 100644 --- a/weed/server/filer_grpc_server_sub_meta.go +++ b/weed/server/filer_grpc_server_sub_meta.go @@ -23,9 +23,12 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest, peerAddress := findClientAddress(stream.Context(), 0) - clientName := fs.addClient(req.ClientName, peerAddress) + alreadyKnown, clientName := fs.addClient(req.ClientName, peerAddress, req.ClientId) + if alreadyKnown { + return fmt.Errorf("duplicated subscription detected for client %s id %d", clientName, req.ClientId) + } - defer fs.deleteClient(clientName) + defer fs.deleteClient(clientName, req.ClientId) lastReadTime := time.Unix(0, req.SinceNs) glog.V(0).Infof(" %v starts to subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime) @@ -81,9 +84,9 @@ func (fs *FilerServer) SubscribeLocalMetadata(req *filer_pb.SubscribeMetadataReq peerAddress := findClientAddress(stream.Context(), 0) - clientName := fs.addClient(req.ClientName, peerAddress) + _, clientName := fs.addClient(req.ClientName, peerAddress, 0) - defer fs.deleteClient(clientName) + defer fs.deleteClient(clientName, 0) lastReadTime := time.Unix(0, req.SinceNs) glog.V(0).Infof(" %v local subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime) @@ -237,12 +240,22 @@ func hasPrefixIn(text string, prefixes []string) bool { return false } -func (fs *FilerServer) addClient(clientType string, clientAddress string) (clientName string) { +func (fs *FilerServer) addClient(clientType string, clientAddress string, clientId int32) (alreadyKnown bool, clientName string) { clientName = clientType + "@" + clientAddress glog.V(0).Infof("+ listener %v", clientName) + if clientId != 0 { + fs.knownListenersLock.Lock() + _, alreadyKnown = fs.knownListeners[clientId] + fs.knownListenersLock.Unlock() + } return } -func (fs *FilerServer) deleteClient(clientName string) { +func (fs *FilerServer) deleteClient(clientName string, clientId int32) { glog.V(0).Infof("- listener %v", clientName) + if clientId != 0 { + fs.knownListenersLock.Lock() + delete(fs.knownListeners, clientId) + fs.knownListenersLock.Unlock() + } } |