replicator.go 2.0 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768
  1. package replication
  2. import (
  3. "context"
  4. "fmt"
  5. "path/filepath"
  6. "strings"
  7. "github.com/chrislusf/seaweedfs/weed/glog"
  8. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  9. "github.com/chrislusf/seaweedfs/weed/replication/sink"
  10. "github.com/chrislusf/seaweedfs/weed/replication/source"
  11. "github.com/chrislusf/seaweedfs/weed/util"
  12. )
  13. type Replicator struct {
  14. sink sink.ReplicationSink
  15. source *source.FilerSource
  16. }
  17. func NewReplicator(sourceConfig util.Configuration, configPrefix string, dataSink sink.ReplicationSink) *Replicator {
  18. source := &source.FilerSource{}
  19. source.Initialize(sourceConfig, configPrefix)
  20. dataSink.SetSourceFiler(source)
  21. return &Replicator{
  22. sink: dataSink,
  23. source: source,
  24. }
  25. }
  26. func (r *Replicator) Replicate(ctx context.Context, key string, message *filer_pb.EventNotification) error {
  27. if !strings.HasPrefix(key, r.source.Dir) {
  28. glog.V(4).Infof("skipping %v outside of %v", key, r.source.Dir)
  29. return nil
  30. }
  31. newKey := filepath.ToSlash(filepath.Join(r.sink.GetSinkToDirectory(), key[len(r.source.Dir):]))
  32. glog.V(3).Infof("replicate %s => %s", key, newKey)
  33. key = newKey
  34. if message.OldEntry != nil && message.NewEntry == nil {
  35. glog.V(4).Infof("deleting %v", key)
  36. return r.sink.DeleteEntry(key, message.OldEntry.IsDirectory, message.DeleteChunks)
  37. }
  38. if message.OldEntry == nil && message.NewEntry != nil {
  39. glog.V(4).Infof("creating %v", key)
  40. return r.sink.CreateEntry(key, message.NewEntry)
  41. }
  42. if message.OldEntry == nil && message.NewEntry == nil {
  43. glog.V(0).Infof("weird message %+v", message)
  44. return nil
  45. }
  46. foundExisting, err := r.sink.UpdateEntry(key, message.OldEntry, message.NewParentPath, message.NewEntry, message.DeleteChunks)
  47. if foundExisting {
  48. glog.V(4).Infof("updated %v", key)
  49. return err
  50. }
  51. err = r.sink.DeleteEntry(key, message.OldEntry.IsDirectory, false)
  52. if err != nil {
  53. return fmt.Errorf("delete old entry %v: %v", key, err)
  54. }
  55. glog.V(4).Infof("creating missing %v", key)
  56. return r.sink.CreateEntry(key, message.NewEntry)
  57. }