filer: able to tail meta data changes

weed/server/filer_grpc_server_listen.go

@@ -1,6 +1,9 @@
 package weed_server
 
 import (
+	"strings"
+	"time"
+
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 )
@@ -9,62 +12,56 @@ func (fs *FilerServer) ListenForEvents(req *filer_pb.ListenForEventsRequest, str
 
 	peerAddress := findClientAddress(stream.Context(), 0)
 
-	clientName, messageChan := fs.addClient(req.ClientName, peerAddress)
+	clientName := fs.addClient(req.ClientName, peerAddress)
 
-	defer fs.deleteClient(clientName, messageChan)
+	defer fs.deleteClient(clientName)
 
-	// ts := time.Unix(req.SinceSec, 0)
+	lastReadTime := time.Now()
+	if req.SinceNs > 0 {
+		lastReadTime = time.Unix(0, req.SinceNs)
+	}
+	var readErr error
+	for {
 
-	// iterate through old messages
-	/*
-		for _, message := range ms.Topo.ToVolumeLocations() {
+		lastReadTime, readErr = fs.filer.ReadLogBuffer(lastReadTime, func(fullpath string, eventNotification *filer_pb.EventNotification) error {
+			if strings.HasPrefix(fullpath, "/.meta") {
+				return nil
+			}
+			if !strings.HasPrefix(fullpath, req.Directory) {
+				return nil
+			}
+			message := &filer_pb.FullEventNotification{
+				Directory:         fullpath,
+				EventNotification: eventNotification,
+			}
 			if err := stream.Send(message); err != nil {
 				return err
 			}
+			return nil
+		})
+		if readErr != nil {
+			glog.V(0).Infof("=> client %v: %+v", clientName, readErr)
+			return readErr
 		}
-	*/
 
-	// need to add a buffer here to avoid slow clients
-	// also needs to support millions of clients
-
-	for message := range messageChan {
-		if err := stream.Send(message); err != nil {
-			glog.V(0).Infof("=> client %v: %+v", clientName, message)
-			return err
-		}
+		fs.listenersLock.Lock()
+		fs.listenersCond.Wait()
+		fs.listenersLock.Unlock()
 	}
 
 	return nil
 }
 
-func (fs *FilerServer) addClient(clientType string, clientAddress string) (clientName string, messageChan chan *filer_pb.FullEventNotification) {
+func (fs *FilerServer) addClient(clientType string, clientAddress string) (clientName string) {
 	clientName = clientType + "@" + clientAddress
 	glog.V(0).Infof("+ listener %v", clientName)
-
-	messageChan = make(chan *filer_pb.FullEventNotification, 10)
-
-	fs.clientChansLock.Lock()
-	fs.clientChans[clientName] = messageChan
-	fs.clientChansLock.Unlock()
 	return
 }
 
-func (fs *FilerServer) deleteClient(clientName string, messageChan chan *filer_pb.FullEventNotification) {
+func (fs *FilerServer) deleteClient(clientName string) {
 	glog.V(0).Infof("- listener %v", clientName)
-	close(messageChan)
-	fs.clientChansLock.Lock()
-	delete(fs.clientChans, clientName)
-	fs.clientChansLock.Unlock()
 }
 
-func (fs *FilerServer) sendMessageToClients(dir string, eventNotification *filer_pb.EventNotification) {
-	message := &filer_pb.FullEventNotification{
-		Directory:         dir,
-		EventNotification: eventNotification,
-	}
-	fs.clientChansLock.RLock()
-	for _, ch := range fs.clientChans {
-		ch <- message
-	}
-	fs.clientChansLock.RUnlock()
+func (fs *FilerServer) notifyMetaListeners() {
+	fs.listenersCond.Broadcast()
 }

weed/server/filer_server.go

@@ -12,7 +12,6 @@ import (
 
 	"github.com/chrislusf/seaweedfs/weed/operation"
 	"github.com/chrislusf/seaweedfs/weed/pb"
-	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
 	"github.com/chrislusf/seaweedfs/weed/stats"
 	"github.com/chrislusf/seaweedfs/weed/util"
@@ -57,8 +56,8 @@ type FilerServer struct {
 	grpcDialOption grpc.DialOption
 
 	// notifying clients
-	clientChansLock sync.RWMutex
-	clientChans     map[string]chan *filer_pb.FullEventNotification
+	listenersLock sync.Mutex
+	listenersCond *sync.Cond
 }
 
 func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption) (fs *FilerServer, err error) {
@@ -67,12 +66,13 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 		option:         option,
 		grpcDialOption: security.LoadClientTLS(util.GetViper(), "grpc.filer"),
 	}
+	fs.listenersCond = sync.NewCond(&fs.listenersLock)
 
 	if len(option.Masters) == 0 {
 		glog.Fatal("master list is required!")
 	}
 
-	fs.filer = filer2.NewFiler(option.Masters, fs.grpcDialOption, option.Port+10000)
+	fs.filer = filer2.NewFiler(option.Masters, fs.grpcDialOption, option.Port+10000, fs.notifyMetaListeners)
 	fs.filer.Cipher = option.Cipher
 
 	maybeStartMetrics(fs, option)