add client id for all metadata listening clients

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()
+	}
 }