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

use only one metadata follow process

Browse Source
This commit is contained in:
chrislu
2024-10-06 12:55:19 -07:00
parent 44b275879b
commit c0e36231ad
3 changed files with 46 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
weed/mount/meta_cache/meta_cache_subscribe.go
View File

@@ -10,7 +10,42 @@ import (
"strings"
)
func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.FilerClient, dir string, lastTsNs int64) error {
type MetadataFollower struct {
PathPrefixToWatch string
ProcessEventFn func(resp *filer_pb.SubscribeMetadataResponse) error
}
func mergeProceesors(mainProcessor func(resp *filer_pb.SubscribeMetadataResponse) error, followers ...*MetadataFollower) func(resp *filer_pb.SubscribeMetadataResponse) error {
return func(resp *filer_pb.SubscribeMetadataResponse) error {
// build the full path
entry := resp.EventNotification.NewEntry
if entry == nil {
entry = resp.EventNotification.OldEntry
}
dir := resp.Directory
if resp.EventNotification.NewParentPath != "" {
dir = resp.EventNotification.NewParentPath
}
fp := util.NewFullPath(dir, entry.Name)
for _, follower := range followers {
if strings.HasPrefix(string(fp), follower.PathPrefixToWatch) {
if err := follower.ProcessEventFn(resp); err != nil {
return err
}
}
}
return mainProcessor(resp)
}
}
func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.FilerClient, dir string, lastTsNs int64, followers ...*MetadataFollower) error {
var prefixes []string
for _, follower := range followers {
prefixes = append(prefixes, follower.PathPrefixToWatch)
}
processEventFn := func(resp *filer_pb.SubscribeMetadataResponse) error {
message := resp.EventNotification
@@ -69,7 +104,7 @@ func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.Fil
ClientEpoch: 1,
SelfSignature: selfSignature,
PathPrefix: prefix,
AdditionalPathPrefixes: nil,
AdditionalPathPrefixes: prefixes,
DirectoriesToWatch: nil,
StartTsNs: lastTsNs,
StopTsNs: 0,
@@ -77,7 +112,7 @@ func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.Fil
}
util.RetryUntil("followMetaUpdates", func() error {
metadataFollowOption.ClientEpoch++
return pb.WithFilerClientFollowMetadata(client, metadataFollowOption, processEventFn)
return pb.WithFilerClientFollowMetadata(client, metadataFollowOption, mergeProceesors(processEventFn, followers...))
}, func(err error) bool {
glog.Errorf("follow metadata updates: %v", err)
return true