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.

143 lines
3.2 KiB

  1. package redis
  2. import (
  3. "context"
  4. "fmt"
  5. "github.com/chrislusf/seaweedfs/weed/filer2"
  6. "github.com/chrislusf/seaweedfs/weed/glog"
  7. "github.com/go-redis/redis"
  8. "sort"
  9. "strings"
  10. "time"
  11. )
  12. const (
  13. DIR_LIST_MARKER = "\x00"
  14. )
  15. type UniversalRedisStore struct {
  16. Client redis.UniversalClient
  17. }
  18. func (store *UniversalRedisStore) InsertEntry(ctx context.Context, entry *filer2.Entry) (err error) {
  19. value, err := entry.EncodeAttributesAndChunks()
  20. if err != nil {
  21. return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err)
  22. }
  23. _, err = store.Client.Set(string(entry.FullPath), value, time.Duration(entry.TtlSec)*time.Second).Result()
  24. if err != nil {
  25. return fmt.Errorf("persisting %s : %v", entry.FullPath, err)
  26. }
  27. dir, name := entry.FullPath.DirAndName()
  28. if name != "" {
  29. _, err = store.Client.SAdd(genDirectoryListKey(dir), name).Result()
  30. if err != nil {
  31. return fmt.Errorf("persisting %s in parent dir: %v", entry.FullPath, err)
  32. }
  33. }
  34. return nil
  35. }
  36. func (store *UniversalRedisStore) UpdateEntry(ctx context.Context, entry *filer2.Entry) (err error) {
  37. return store.InsertEntry(ctx, entry)
  38. }
  39. func (store *UniversalRedisStore) FindEntry(ctx context.Context, fullpath filer2.FullPath) (entry *filer2.Entry, err error) {
  40. data, err := store.Client.Get(string(fullpath)).Result()
  41. if err == redis.Nil {
  42. return nil, filer2.ErrNotFound
  43. }
  44. if err != nil {
  45. return nil, fmt.Errorf("get %s : %v", fullpath, err)
  46. }
  47. entry = &filer2.Entry{
  48. FullPath: fullpath,
  49. }
  50. err = entry.DecodeAttributesAndChunks([]byte(data))
  51. if err != nil {
  52. return entry, fmt.Errorf("decode %s : %v", entry.FullPath, err)
  53. }
  54. return entry, nil
  55. }
  56. func (store *UniversalRedisStore) DeleteEntry(ctx context.Context, fullpath filer2.FullPath) (err error) {
  57. _, err = store.Client.Del(string(fullpath)).Result()
  58. if err != nil {
  59. return fmt.Errorf("delete %s : %v", fullpath, err)
  60. }
  61. dir, name := fullpath.DirAndName()
  62. if name != "" {
  63. _, err = store.Client.SRem(genDirectoryListKey(dir), name).Result()
  64. if err != nil {
  65. return fmt.Errorf("delete %s in parent dir: %v", fullpath, err)
  66. }
  67. }
  68. return nil
  69. }
  70. func (store *UniversalRedisStore) ListDirectoryEntries(ctx context.Context, fullpath filer2.FullPath, startFileName string, inclusive bool,
  71. limit int) (entries []*filer2.Entry, err error) {
  72. members, err := store.Client.SMembers(genDirectoryListKey(string(fullpath))).Result()
  73. if err != nil {
  74. return nil, fmt.Errorf("list %s : %v", fullpath, err)
  75. }
  76. // skip
  77. if startFileName != "" {
  78. var t []string
  79. for _, m := range members {
  80. if strings.Compare(m, startFileName) >= 0 {
  81. if m == startFileName {
  82. if inclusive {
  83. t = append(t, m)
  84. }
  85. } else {
  86. t = append(t, m)
  87. }
  88. }
  89. }
  90. members = t
  91. }
  92. // sort
  93. sort.Slice(members, func(i, j int) bool {
  94. return strings.Compare(members[i], members[j]) < 0
  95. })
  96. // limit
  97. if limit < len(members) {
  98. members = members[:limit]
  99. }
  100. // fetch entry meta
  101. for _, fileName := range members {
  102. path := filer2.NewFullPath(string(fullpath), fileName)
  103. entry, err := store.FindEntry(ctx, path)
  104. if err != nil {
  105. glog.V(0).Infof("list %s : %v", path, err)
  106. } else {
  107. entries = append(entries, entry)
  108. }
  109. }
  110. return entries, err
  111. }
  112. func genDirectoryListKey(dir string) (dirList string) {
  113. return dir + DIR_LIST_MARKER
  114. }