watch entries with common path prefix

1 files changed, 18 insertions, 3 deletions

diff --git a/weed/server/filer_grpc_server_listen.go b/weed/server/filer_grpc_server_listen.go
index 81b0a9254..a3df1cf9d 100644
--- a/weed/server/filer_grpc_server_listen.go
+++ b/weed/server/filer_grpc_server_listen.go
@@ -6,6 +6,7 @@ import (
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
+	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
 func (fs *FilerServer) ListenForEvents(req *filer_pb.ListenForEventsRequest, stream filer_pb.SeaweedFiler_ListenForEventsServer) error {
@@ -23,15 +24,29 @@ func (fs *FilerServer) ListenForEvents(req *filer_pb.ListenForEventsRequest, str
 	var readErr error
 	for {
 
-		lastReadTime, readErr = fs.filer.ReadLogBuffer(lastReadTime, func(fullpath string, eventNotification *filer_pb.EventNotification) error {
+		lastReadTime, readErr = fs.filer.ReadLogBuffer(lastReadTime, func(dirPath string, eventNotification *filer_pb.EventNotification) error {
+
+			// get complete path to the file or directory
+			var entryName string
+			if eventNotification.OldEntry != nil {
+				entryName = eventNotification.OldEntry.Name
+			} else if eventNotification.NewEntry != nil {
+				entryName = eventNotification.NewEntry.Name
+			}
+
+			fullpath := util.PathJoin(dirPath, entryName)
+
+			// skip on filer internal meta logs
 			if strings.HasPrefix(fullpath, "/.meta") {
 				return nil
 			}
-			if !strings.HasPrefix(fullpath, req.Directory) {
+
+			if !strings.HasPrefix(fullpath, req.PathPrefix) {
 				return nil
 			}
+
 			message := &filer_pb.FullEventNotification{
-				Directory:         fullpath,
+				Directory:         dirPath,
 				EventNotification: eventNotification,
 			}
 			if err := stream.Send(message); err != nil {