weedfs_write.go 1.7 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162
  1. package mount
  2. import (
  3. "fmt"
  4. "io"
  5. "github.com/seaweedfs/seaweedfs/weed/filer"
  6. "github.com/seaweedfs/seaweedfs/weed/glog"
  7. "github.com/seaweedfs/seaweedfs/weed/operation"
  8. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  9. "github.com/seaweedfs/seaweedfs/weed/util"
  10. )
  11. func (wfs *WFS) saveDataAsChunk(fullPath util.FullPath) filer.SaveDataAsChunkFunctionType {
  12. return func(reader io.Reader, filename string, offset int64) (chunk *filer_pb.FileChunk, err error) {
  13. fileId, uploadResult, err, data := operation.UploadWithRetry(
  14. wfs,
  15. &filer_pb.AssignVolumeRequest{
  16. Count: 1,
  17. Replication: wfs.option.Replication,
  18. Collection: wfs.option.Collection,
  19. TtlSec: wfs.option.TtlSec,
  20. DiskType: string(wfs.option.DiskType),
  21. DataCenter: wfs.option.DataCenter,
  22. Path: string(fullPath),
  23. },
  24. &operation.UploadOption{
  25. Filename: filename,
  26. Cipher: wfs.option.Cipher,
  27. IsInputCompressed: false,
  28. MimeType: "",
  29. PairMap: nil,
  30. },
  31. func(host, fileId string) string {
  32. fileUrl := fmt.Sprintf("http://%s/%s", host, fileId)
  33. if wfs.option.VolumeServerAccess == "filerProxy" {
  34. fileUrl = fmt.Sprintf("http://%s/?proxyChunkId=%s", wfs.getCurrentFiler(), fileId)
  35. }
  36. return fileUrl
  37. },
  38. reader,
  39. )
  40. if err != nil {
  41. glog.V(0).Infof("upload data %v: %v", filename, err)
  42. return nil, fmt.Errorf("upload data: %v", err)
  43. }
  44. if uploadResult.Error != "" {
  45. glog.V(0).Infof("upload failure %v: %v", filename, err)
  46. return nil, fmt.Errorf("upload result: %v", uploadResult.Error)
  47. }
  48. if offset == 0 {
  49. wfs.chunkCache.SetChunk(fileId, data)
  50. }
  51. chunk = uploadResult.ToPbFileChunk(fileId, offset)
  52. return chunk, nil
  53. }
  54. }