diff options
Diffstat (limited to 'weed')
| -rw-r--r-- | weed/server/filer_grpc_server_listen.go | 8 |
1 files changed, 8 insertions, 0 deletions
diff --git a/weed/server/filer_grpc_server_listen.go b/weed/server/filer_grpc_server_listen.go index 2a18f1950..e45068f39 100644 --- a/weed/server/filer_grpc_server_listen.go +++ b/weed/server/filer_grpc_server_listen.go @@ -22,6 +22,8 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest, defer fs.deleteClient(clientName) lastReadTime := time.Unix(0, req.SinceNs) + glog.V(0).Infof(" %v starts to subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime) + var processedTsNs int64 eachEventNotificationFn := func(dirPath string, eventNotification *filer_pb.EventNotification, tsNs int64) error { @@ -67,6 +69,8 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest, return err } + processedTsNs = logEntry.TsNs + return nil } @@ -74,6 +78,10 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest, return fmt.Errorf("reading from persisted logs: %v", err) } + if processedTsNs != 0 { + lastReadTime = time.Unix(0, processedTsNs) + } + _, err := fs.filer.MetaLogBuffer.LoopProcessLogData(lastReadTime, func() bool { fs.listenersLock.Lock() fs.listenersCond.Wait() |