filer_pb_helper.go 3.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149
  1. package filer_pb
  2. import (
  3. "context"
  4. "errors"
  5. "fmt"
  6. "strings"
  7. "github.com/chrislusf/seaweedfs/weed/glog"
  8. "github.com/chrislusf/seaweedfs/weed/storage/needle"
  9. "github.com/golang/protobuf/proto"
  10. "github.com/viant/ptrie"
  11. )
  12. func ToFileIdObject(fileIdStr string) (*FileId, error) {
  13. t, err := needle.ParseFileIdFromString(fileIdStr)
  14. if err != nil {
  15. return nil, err
  16. }
  17. return &FileId{
  18. VolumeId: uint32(t.VolumeId),
  19. Cookie: uint32(t.Cookie),
  20. FileKey: uint64(t.Key),
  21. }, nil
  22. }
  23. func (fid *FileId) toFileIdString() string {
  24. return needle.NewFileId(needle.VolumeId(fid.VolumeId), fid.FileKey, fid.Cookie).String()
  25. }
  26. func (c *FileChunk) GetFileIdString() string {
  27. if c.FileId != "" {
  28. return c.FileId
  29. }
  30. if c.Fid != nil {
  31. c.FileId = c.Fid.toFileIdString()
  32. return c.FileId
  33. }
  34. return ""
  35. }
  36. func BeforeEntrySerialization(chunks []*FileChunk) {
  37. for _, chunk := range chunks {
  38. if chunk.FileId != "" {
  39. if fid, err := ToFileIdObject(chunk.FileId); err == nil {
  40. chunk.Fid = fid
  41. chunk.FileId = ""
  42. }
  43. }
  44. if chunk.SourceFileId != "" {
  45. if fid, err := ToFileIdObject(chunk.SourceFileId); err == nil {
  46. chunk.SourceFid = fid
  47. chunk.SourceFileId = ""
  48. }
  49. }
  50. }
  51. }
  52. func EnsureFid(chunk *FileChunk) {
  53. if chunk.Fid != nil {
  54. return
  55. }
  56. if fid, err := ToFileIdObject(chunk.FileId); err == nil {
  57. chunk.Fid = fid
  58. }
  59. }
  60. func AfterEntryDeserialization(chunks []*FileChunk) {
  61. for _, chunk := range chunks {
  62. if chunk.Fid != nil && chunk.FileId == "" {
  63. chunk.FileId = chunk.Fid.toFileIdString()
  64. }
  65. if chunk.SourceFid != nil && chunk.SourceFileId == "" {
  66. chunk.SourceFileId = chunk.SourceFid.toFileIdString()
  67. }
  68. }
  69. }
  70. func CreateEntry(client SeaweedFilerClient, request *CreateEntryRequest) error {
  71. resp, err := client.CreateEntry(context.Background(), request)
  72. if err != nil {
  73. glog.V(1).Infof("create entry %s/%s %v: %v", request.Directory, request.Entry.Name, request.OExcl, err)
  74. return fmt.Errorf("CreateEntry: %v", err)
  75. }
  76. if resp.Error != "" {
  77. glog.V(1).Infof("create entry %s/%s %v: %v", request.Directory, request.Entry.Name, request.OExcl, resp.Error)
  78. return fmt.Errorf("CreateEntry : %v", resp.Error)
  79. }
  80. return nil
  81. }
  82. func UpdateEntry(client SeaweedFilerClient, request *UpdateEntryRequest) error {
  83. _, err := client.UpdateEntry(context.Background(), request)
  84. if err != nil {
  85. glog.V(1).Infof("update entry %s/%s :%v", request.Directory, request.Entry.Name, err)
  86. return fmt.Errorf("UpdateEntry: %v", err)
  87. }
  88. return nil
  89. }
  90. func LookupEntry(client SeaweedFilerClient, request *LookupDirectoryEntryRequest) (*LookupDirectoryEntryResponse, error) {
  91. resp, err := client.LookupDirectoryEntry(context.Background(), request)
  92. if err != nil {
  93. if err == ErrNotFound || strings.Contains(err.Error(), ErrNotFound.Error()) {
  94. return nil, ErrNotFound
  95. }
  96. glog.V(3).Infof("read %s/%v: %v", request.Directory, request.Name, err)
  97. return nil, fmt.Errorf("LookupEntry1: %v", err)
  98. }
  99. if resp.Entry == nil {
  100. return nil, ErrNotFound
  101. }
  102. return resp, nil
  103. }
  104. var ErrNotFound = errors.New("filer: no entry is found in filer store")
  105. func IsCreate(event *SubscribeMetadataResponse) bool {
  106. return event.EventNotification.NewEntry != nil && event.EventNotification.OldEntry == nil
  107. }
  108. func IsUpdate(event *SubscribeMetadataResponse) bool {
  109. return event.EventNotification.NewEntry != nil &&
  110. event.EventNotification.OldEntry != nil &&
  111. event.Directory == event.EventNotification.NewParentPath
  112. }
  113. func IsDelete(event *SubscribeMetadataResponse) bool {
  114. return event.EventNotification.NewEntry == nil && event.EventNotification.OldEntry != nil
  115. }
  116. func IsRename(event *SubscribeMetadataResponse) bool {
  117. return event.EventNotification.NewEntry != nil &&
  118. event.EventNotification.OldEntry != nil &&
  119. event.Directory != event.EventNotification.NewParentPath
  120. }
  121. var _ = ptrie.KeyProvider(&FilerConf_PathConf{})
  122. func (fp *FilerConf_PathConf) Key() interface{} {
  123. key, _ := proto.Marshal(fp)
  124. return string(key)
  125. }