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.

223 lines
5.4 KiB

5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
  1. package filer_pb
  2. import (
  3. "context"
  4. "fmt"
  5. "io"
  6. "math"
  7. "os"
  8. "time"
  9. "github.com/chrislusf/seaweedfs/weed/glog"
  10. "github.com/chrislusf/seaweedfs/weed/util"
  11. )
  12. var (
  13. OS_UID = uint32(os.Getuid())
  14. OS_GID = uint32(os.Getgid())
  15. )
  16. type FilerClient interface {
  17. WithFilerClient(fn func(SeaweedFilerClient) error) error
  18. AdjustedUrl(hostAndPort string) string
  19. }
  20. func GetEntry(filerClient FilerClient, fullFilePath util.FullPath) (entry *Entry, err error) {
  21. dir, name := fullFilePath.DirAndName()
  22. err = filerClient.WithFilerClient(func(client SeaweedFilerClient) error {
  23. request := &LookupDirectoryEntryRequest{
  24. Directory: dir,
  25. Name: name,
  26. }
  27. // glog.V(3).Infof("read %s request: %v", fullFilePath, request)
  28. resp, err := LookupEntry(client, request)
  29. if err != nil {
  30. if err == ErrNotFound {
  31. return nil
  32. }
  33. glog.V(3).Infof("read %s %v: %v", fullFilePath, resp, err)
  34. return err
  35. }
  36. if resp.Entry == nil {
  37. // glog.V(3).Infof("read %s entry: %v", fullFilePath, entry)
  38. return nil
  39. }
  40. entry = resp.Entry
  41. return nil
  42. })
  43. return
  44. }
  45. func ReadDirAllEntries(filerClient FilerClient, fullDirPath util.FullPath, prefix string, fn func(entry *Entry, isLast bool)) (err error) {
  46. return doList(filerClient, fullDirPath, prefix, fn, "", false, math.MaxUint32)
  47. }
  48. func List(filerClient FilerClient, parentDirectoryPath, prefix string, fn func(entry *Entry, isLast bool), startFrom string, inclusive bool, limit uint32) (err error) {
  49. return doList(filerClient, util.FullPath(parentDirectoryPath), prefix, fn, startFrom, inclusive, limit)
  50. }
  51. func doList(filerClient FilerClient, fullDirPath util.FullPath, prefix string, fn func(entry *Entry, isLast bool), startFrom string, inclusive bool, limit uint32) (err error) {
  52. err = filerClient.WithFilerClient(func(client SeaweedFilerClient) error {
  53. request := &ListEntriesRequest{
  54. Directory: string(fullDirPath),
  55. Prefix: prefix,
  56. StartFromFileName: startFrom,
  57. Limit: limit,
  58. InclusiveStartFrom: inclusive,
  59. }
  60. glog.V(3).Infof("read directory: %v", request)
  61. stream, err := client.ListEntries(context.Background(), request)
  62. if err != nil {
  63. return fmt.Errorf("list %s: %v", fullDirPath, err)
  64. }
  65. var prevEntry *Entry
  66. for {
  67. resp, recvErr := stream.Recv()
  68. if recvErr != nil {
  69. if recvErr == io.EOF {
  70. if prevEntry != nil {
  71. fn(prevEntry, true)
  72. }
  73. break
  74. } else {
  75. return recvErr
  76. }
  77. }
  78. if prevEntry != nil {
  79. fn(prevEntry, false)
  80. }
  81. prevEntry = resp.Entry
  82. }
  83. return nil
  84. })
  85. return
  86. }
  87. func Exists(filerClient FilerClient, parentDirectoryPath string, entryName string, isDirectory bool) (exists bool, err error) {
  88. err = filerClient.WithFilerClient(func(client SeaweedFilerClient) error {
  89. request := &LookupDirectoryEntryRequest{
  90. Directory: parentDirectoryPath,
  91. Name: entryName,
  92. }
  93. glog.V(4).Infof("exists entry %v/%v: %v", parentDirectoryPath, entryName, request)
  94. resp, err := LookupEntry(client, request)
  95. if err != nil {
  96. if err == ErrNotFound {
  97. exists = false
  98. return nil
  99. }
  100. glog.V(0).Infof("exists entry %v: %v", request, err)
  101. return fmt.Errorf("exists entry %s/%s: %v", parentDirectoryPath, entryName, err)
  102. }
  103. exists = resp.Entry.IsDirectory == isDirectory
  104. return nil
  105. })
  106. return
  107. }
  108. func Mkdir(filerClient FilerClient, parentDirectoryPath string, dirName string, fn func(entry *Entry)) error {
  109. return filerClient.WithFilerClient(func(client SeaweedFilerClient) error {
  110. entry := &Entry{
  111. Name: dirName,
  112. IsDirectory: true,
  113. Attributes: &FuseAttributes{
  114. Mtime: time.Now().Unix(),
  115. Crtime: time.Now().Unix(),
  116. FileMode: uint32(0777 | os.ModeDir),
  117. Uid: OS_UID,
  118. Gid: OS_GID,
  119. },
  120. }
  121. if fn != nil {
  122. fn(entry)
  123. }
  124. request := &CreateEntryRequest{
  125. Directory: parentDirectoryPath,
  126. Entry: entry,
  127. }
  128. glog.V(1).Infof("mkdir: %v", request)
  129. if err := CreateEntry(client, request); err != nil {
  130. glog.V(0).Infof("mkdir %v: %v", request, err)
  131. return fmt.Errorf("mkdir %s/%s: %v", parentDirectoryPath, dirName, err)
  132. }
  133. return nil
  134. })
  135. }
  136. func MkFile(filerClient FilerClient, parentDirectoryPath string, fileName string, chunks []*FileChunk) error {
  137. return filerClient.WithFilerClient(func(client SeaweedFilerClient) error {
  138. entry := &Entry{
  139. Name: fileName,
  140. IsDirectory: false,
  141. Attributes: &FuseAttributes{
  142. Mtime: time.Now().Unix(),
  143. Crtime: time.Now().Unix(),
  144. FileMode: uint32(0770),
  145. Uid: OS_UID,
  146. Gid: OS_GID,
  147. },
  148. Chunks: chunks,
  149. }
  150. request := &CreateEntryRequest{
  151. Directory: parentDirectoryPath,
  152. Entry: entry,
  153. }
  154. glog.V(1).Infof("create file: %s/%s", parentDirectoryPath, fileName)
  155. if err := CreateEntry(client, request); err != nil {
  156. glog.V(0).Infof("create file %v:%v", request, err)
  157. return fmt.Errorf("create file %s/%s: %v", parentDirectoryPath, fileName, err)
  158. }
  159. return nil
  160. })
  161. }
  162. func Remove(filerClient FilerClient, parentDirectoryPath string, name string, isDeleteData, isRecursive, ignoreRecursiveErr bool) error {
  163. return filerClient.WithFilerClient(func(client SeaweedFilerClient) error {
  164. if _, err := client.DeleteEntry(context.Background(), &DeleteEntryRequest{
  165. Directory: parentDirectoryPath,
  166. Name: name,
  167. IsDeleteData: isDeleteData,
  168. IsRecursive: isRecursive,
  169. IgnoreRecursiveError: ignoreRecursiveErr,
  170. }); err != nil {
  171. return err
  172. }
  173. return nil
  174. })
  175. }