command_fs_meta_notify.go 1.7 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677
  1. package shell
  2. import (
  3. "fmt"
  4. "io"
  5. "github.com/seaweedfs/seaweedfs/weed/notification"
  6. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  7. "github.com/seaweedfs/seaweedfs/weed/util"
  8. )
  9. func init() {
  10. Commands = append(Commands, &commandFsMetaNotify{})
  11. }
  12. type commandFsMetaNotify struct {
  13. }
  14. func (c *commandFsMetaNotify) Name() string {
  15. return "fs.meta.notify"
  16. }
  17. func (c *commandFsMetaNotify) Help() string {
  18. return `recursively send directory and file meta data to notification message queue
  19. fs.meta.notify # send meta data from current directory to notification message queue
  20. The message queue will use it to trigger replication from this filer.
  21. `
  22. }
  23. func (c *commandFsMetaNotify) HasTag(CommandTag) bool {
  24. return false
  25. }
  26. func (c *commandFsMetaNotify) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
  27. path, err := commandEnv.parseUrl(findInputDirectory(args))
  28. if err != nil {
  29. return err
  30. }
  31. util.LoadConfiguration("notification", true)
  32. v := util.GetViper()
  33. notification.LoadConfiguration(v, "notification.")
  34. var dirCount, fileCount uint64
  35. err = filer_pb.TraverseBfs(commandEnv, util.FullPath(path), func(parentPath util.FullPath, entry *filer_pb.Entry) {
  36. if entry.IsDirectory {
  37. dirCount++
  38. } else {
  39. fileCount++
  40. }
  41. notifyErr := notification.Queue.SendMessage(
  42. string(parentPath.Child(entry.Name)),
  43. &filer_pb.EventNotification{
  44. NewEntry: entry,
  45. },
  46. )
  47. if notifyErr != nil {
  48. fmt.Fprintf(writer, "fail to notify new entry event for %s: %v\n", parentPath.Child(entry.Name), notifyErr)
  49. }
  50. })
  51. if err == nil {
  52. fmt.Fprintf(writer, "\ntotal notified %d directories, %d files\n", dirCount, fileCount)
  53. }
  54. return err
  55. }