You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

68 lines
2.0 KiB

  1. package meta_cache
  2. import (
  3. "context"
  4. "github.com/chrislusf/seaweedfs/weed/filer"
  5. "github.com/chrislusf/seaweedfs/weed/glog"
  6. "github.com/chrislusf/seaweedfs/weed/pb"
  7. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  8. "github.com/chrislusf/seaweedfs/weed/util"
  9. )
  10. func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.FilerClient, dir string, lastTsNs int64) error {
  11. processEventFn := func(resp *filer_pb.SubscribeMetadataResponse) error {
  12. message := resp.EventNotification
  13. for _, sig := range message.Signatures {
  14. if sig == selfSignature && selfSignature != 0 {
  15. return nil
  16. }
  17. }
  18. dir := resp.Directory
  19. var oldPath util.FullPath
  20. var newEntry *filer.Entry
  21. if message.OldEntry != nil {
  22. oldPath = util.NewFullPath(dir, message.OldEntry.Name)
  23. glog.V(4).Infof("deleting %v", oldPath)
  24. }
  25. if message.NewEntry != nil {
  26. if message.NewParentPath != "" {
  27. dir = message.NewParentPath
  28. }
  29. key := util.NewFullPath(dir, message.NewEntry.Name)
  30. glog.V(4).Infof("creating %v", key)
  31. newEntry = filer.FromPbEntry(dir, message.NewEntry)
  32. }
  33. err := mc.AtomicUpdateEntryFromFiler(context.Background(), oldPath, newEntry)
  34. if err == nil {
  35. if message.OldEntry != nil && message.NewEntry != nil {
  36. oldKey := util.NewFullPath(resp.Directory, message.OldEntry.Name)
  37. mc.invalidateFunc(oldKey)
  38. if message.OldEntry.Name != message.NewEntry.Name {
  39. newKey := util.NewFullPath(dir, message.NewEntry.Name)
  40. mc.invalidateFunc(newKey)
  41. }
  42. } else if message.OldEntry == nil && message.NewEntry != nil {
  43. // no need to invaalidate
  44. } else if message.OldEntry != nil && message.NewEntry == nil {
  45. oldKey := util.NewFullPath(resp.Directory, message.OldEntry.Name)
  46. mc.invalidateFunc(oldKey)
  47. }
  48. }
  49. return err
  50. }
  51. util.RetryForever("followMetaUpdates", func() error {
  52. return pb.WithFilerClientFollowMetadata(client, "mount", dir, lastTsNs, selfSignature, processEventFn, true)
  53. }, func(err error) bool {
  54. glog.Errorf("follow metadata updates: %v", err)
  55. return true
  56. })
  57. return nil
  58. }