gsh-digital-services/seaweedFS
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

FUSE: subcribe meta events based on timestamp

Browse Source
This commit is contained in:
Chris Lu
2020-04-27 16:52:20 -07:00
parent 6190fd665d
commit 2a7957b4ca
2 changed files with 3 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
weed/filesys/meta_cache/meta_cache_subscribe.go
View File

@@ -12,7 +12,7 @@ import (
"github.com/chrislusf/seaweedfs/weed/util" "github.com/chrislusf/seaweedfs/weed/util"
) )
func SubscribeMetaEvents(mc *MetaCache, client filer_pb.FilerClient, dir string) error { func SubscribeMetaEvents(mc *MetaCache, client filer_pb.FilerClient, dir string, lastTsNs int64) error {
processEventFn := func(resp *filer_pb.SubscribeMetadataResponse) error { processEventFn := func(resp *filer_pb.SubscribeMetadataResponse) error {
message := resp.EventNotification message := resp.EventNotification
@@ -36,7 +36,6 @@ func SubscribeMetaEvents(mc *MetaCache, client filer_pb.FilerClient, dir string)
return err return err
} }
var lastTsNs int64
for { for {
err := client.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error { err := client.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
stream, err := client.SubscribeMetadata(context.Background(), &filer_pb.SubscribeMetadataRequest{ stream, err := client.SubscribeMetadata(context.Background(), &filer_pb.SubscribeMetadataRequest{

3
weed/filesys/wfs.go
View File

@@ -95,10 +95,11 @@ func NewSeaweedFileSystem(option *Option) *WFS {
} }
if wfs.option.AsyncMetaDataCaching { if wfs.option.AsyncMetaDataCaching {
wfs.metaCache = meta_cache.NewMetaCache(path.Join(option.CacheDir, "meta")) wfs.metaCache = meta_cache.NewMetaCache(path.Join(option.CacheDir, "meta"))
startTime := time.Now()
if err := meta_cache.InitMetaCache(wfs.metaCache, wfs, wfs.option.FilerMountRootPath); err != nil{ if err := meta_cache.InitMetaCache(wfs.metaCache, wfs, wfs.option.FilerMountRootPath); err != nil{
glog.V(0).Infof("failed to init meta cache: %v", err) glog.V(0).Infof("failed to init meta cache: %v", err)
} else { } else {
go meta_cache.SubscribeMetaEvents(wfs.metaCache, wfs, wfs.option.FilerMountRootPath) go meta_cache.SubscribeMetaEvents(wfs.metaCache, wfs, wfs.option.FilerMountRootPath, startTime.UnixNano())
util.OnInterrupt(func() { util.OnInterrupt(func() {
wfs.metaCache.Shutdown() wfs.metaCache.Shutdown()
}) })