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.

182 lines
4.2 KiB

7 years ago
7 years ago
  1. package filesys
  2. import (
  3. "context"
  4. "fmt"
  5. "bazil.org/fuse"
  6. "bazil.org/fuse/fs"
  7. "github.com/chrislusf/seaweedfs/weed/glog"
  8. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  9. "path/filepath"
  10. "os"
  11. "time"
  12. "bytes"
  13. "github.com/chrislusf/seaweedfs/weed/operation"
  14. )
  15. var _ = fs.Node(&File{})
  16. // var _ = fs.NodeOpener(&File{})
  17. var _ = fs.NodeFsyncer(&File{})
  18. var _ = fs.Handle(&File{})
  19. var _ = fs.HandleReadAller(&File{})
  20. // var _ = fs.HandleReader(&File{})
  21. var _ = fs.HandleFlusher(&File{})
  22. var _ = fs.HandleWriter(&File{})
  23. type File struct {
  24. Chunks []*filer_pb.FileChunk
  25. Name string
  26. dir *Dir
  27. wfs *WFS
  28. }
  29. func (file *File) Attr(context context.Context, attr *fuse.Attr) error {
  30. fullPath := filepath.Join(file.dir.Path, file.Name)
  31. item := file.wfs.listDirectoryEntriesCache.Get(fullPath)
  32. var attributes *filer_pb.FuseAttributes
  33. if item != nil {
  34. attributes = item.Value().(*filer_pb.FuseAttributes)
  35. glog.V(1).Infof("read cached file %v attributes", file.Name)
  36. } else {
  37. err := file.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  38. request := &filer_pb.GetFileAttributesRequest{
  39. Name: file.Name,
  40. ParentDir: file.dir.Path,
  41. }
  42. glog.V(1).Infof("read file size: %v", request)
  43. resp, err := client.GetFileAttributes(context, request)
  44. if err != nil {
  45. return err
  46. }
  47. attributes = resp.Attributes
  48. return nil
  49. })
  50. if err != nil {
  51. return err
  52. }
  53. }
  54. attr.Mode = os.FileMode(attributes.FileMode)
  55. attr.Size = attributes.FileSize
  56. attr.Mtime = time.Unix(attributes.Mtime, 0)
  57. attr.Gid = attributes.Gid
  58. attr.Uid = attributes.Uid
  59. return nil
  60. }
  61. func (file *File) ReadAll(ctx context.Context) (content []byte, err error) {
  62. if len(file.Chunks) == 0 {
  63. return
  64. }
  65. err = file.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  66. // FIXME: need to either use Read() or implement differently
  67. request := &filer_pb.GetFileContentRequest{
  68. FileId: file.Chunks[0].FileId,
  69. }
  70. glog.V(1).Infof("read file content: %v", request)
  71. resp, err := client.GetFileContent(ctx, request)
  72. if err != nil {
  73. return err
  74. }
  75. content = resp.Content
  76. return nil
  77. })
  78. return content, err
  79. }
  80. func (file *File) Fsync(ctx context.Context, req *fuse.FsyncRequest) error {
  81. // fsync works at OS level
  82. // write the file chunks to the filer
  83. fmt.Printf("flush file %+v\n", req)
  84. return nil
  85. }
  86. func (file *File) Flush(ctx context.Context, req *fuse.FlushRequest) error {
  87. // fflush works at file level
  88. // send the data to the OS
  89. fmt.Printf("flush file %+v\n", req)
  90. err := file.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  91. request := &filer_pb.AppendFileChunksRequest{
  92. Directory: file.dir.Path,
  93. Entry: &filer_pb.Entry{
  94. Name: file.Name,
  95. Chunks: file.Chunks,
  96. },
  97. }
  98. glog.V(1).Infof("append chunks: %v", request)
  99. if _, err := client.AppendFileChunks(ctx, request); err != nil {
  100. return fmt.Errorf("create file: %v", err)
  101. }
  102. return nil
  103. })
  104. return err
  105. }
  106. func (file *File) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.WriteResponse) error {
  107. // write the request to volume servers
  108. fmt.Printf("write file %+v\n", req)
  109. var fileId, host string
  110. if err := file.wfs.withFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  111. request := &filer_pb.AssignVolumeRequest{
  112. Count: 1,
  113. Replication: "000",
  114. Collection: "",
  115. }
  116. glog.V(1).Infof("assign volume: %v", request)
  117. resp, err := client.AssignVolume(ctx, request)
  118. if err != nil {
  119. return err
  120. }
  121. fileId, host = resp.FileId, resp.Url
  122. return nil
  123. }); err != nil {
  124. return fmt.Errorf("filer assign volume: %v", err)
  125. }
  126. fileUrl := fmt.Sprintf("http://%s/%s", host, fileId)
  127. bufReader := bytes.NewReader(req.Data)
  128. uploadResult, err := operation.Upload(fileUrl, file.Name, bufReader, false, "application/octet-stream", nil, "")
  129. if err != nil {
  130. return fmt.Errorf("upload data: %v", err)
  131. }
  132. if uploadResult.Error != "" {
  133. return fmt.Errorf("upload result: %v", uploadResult.Error)
  134. }
  135. glog.V(1).Infof("uploaded %s/%s to: %v", file.dir.Path, file.Name, fileUrl)
  136. file.Chunks = append(file.Chunks, &filer_pb.FileChunk{
  137. FileId: fileId,
  138. Offset: req.Offset,
  139. Size: uint64(uploadResult.Size),
  140. Mtime: time.Now().UnixNano(),
  141. })
  142. return nil
  143. }