diff options
Diffstat (limited to 'weed/server')
| -rw-r--r-- | weed/server/filer_grpc_server_listen.go | 71 | ||||
| -rw-r--r-- | weed/server/filer_server.go | 8 |
2 files changed, 38 insertions, 41 deletions
diff --git a/weed/server/filer_grpc_server_listen.go b/weed/server/filer_grpc_server_listen.go index 37d643ce2..81b0a9254 100644 --- a/weed/server/filer_grpc_server_listen.go +++ b/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() } diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go index ba7b0ed03..22bbd2a03 100644 --- a/weed/server/filer_server.go +++ b/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) |