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.

113 lines
3.1 KiB

  1. package broker
  2. import (
  3. "context"
  4. "fmt"
  5. "time"
  6. "github.com/chrislusf/seaweedfs/weed/glog"
  7. "github.com/chrislusf/seaweedfs/weed/operation"
  8. "github.com/chrislusf/seaweedfs/weed/pb"
  9. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  10. "github.com/chrislusf/seaweedfs/weed/pb/messaging_pb"
  11. "github.com/chrislusf/seaweedfs/weed/security"
  12. "github.com/chrislusf/seaweedfs/weed/util"
  13. )
  14. func (broker *MessageBroker) appendToFile(targetFile string, topicConfig *messaging_pb.TopicConfiguration, data []byte) error {
  15. assignResult, uploadResult, err2 := broker.assignAndUpload(topicConfig, data)
  16. if err2 != nil {
  17. return err2
  18. }
  19. dir, name := util.FullPath(targetFile).DirAndName()
  20. chunk := &filer_pb.FileChunk{
  21. FileId: assignResult.Fid,
  22. Offset: 0, // needs to be fixed during appending
  23. Size: uint64(uploadResult.Size),
  24. Mtime: time.Now().UnixNano(),
  25. ETag: uploadResult.ETag,
  26. IsGzipped: uploadResult.Gzip > 0,
  27. }
  28. // append the chunk
  29. if err := broker.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  30. request := &filer_pb.AppendToEntryRequest{
  31. Directory: dir,
  32. EntryName: name,
  33. Chunks: []*filer_pb.FileChunk{chunk},
  34. }
  35. _, err := client.AppendToEntry(context.Background(), request)
  36. if err != nil {
  37. glog.V(0).Infof("append to file %v: %v", request, err)
  38. return err
  39. }
  40. return nil
  41. }); err != nil {
  42. return fmt.Errorf("append to file %v: %v", targetFile, err)
  43. }
  44. return nil
  45. }
  46. func (broker *MessageBroker) assignAndUpload(topicConfig *messaging_pb.TopicConfiguration, data []byte) (*operation.AssignResult, *operation.UploadResult, error) {
  47. var assignResult = &operation.AssignResult{}
  48. // assign a volume location
  49. if err := broker.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
  50. request := &filer_pb.AssignVolumeRequest{
  51. Count: 1,
  52. Replication: topicConfig.Replication,
  53. Collection: topicConfig.Collection,
  54. }
  55. resp, err := client.AssignVolume(context.Background(), request)
  56. if err != nil {
  57. glog.V(0).Infof("assign volume failure %v: %v", request, err)
  58. return err
  59. }
  60. if resp.Error != "" {
  61. return fmt.Errorf("assign volume failure %v: %v", request, resp.Error)
  62. }
  63. assignResult.Auth = security.EncodedJwt(resp.Auth)
  64. assignResult.Fid = resp.FileId
  65. assignResult.Url = resp.Url
  66. assignResult.PublicUrl = resp.PublicUrl
  67. assignResult.Count = uint64(resp.Count)
  68. return nil
  69. }); err != nil {
  70. return nil, nil, err
  71. }
  72. // upload data
  73. targetUrl := fmt.Sprintf("http://%s/%s", assignResult.Url, assignResult.Fid)
  74. uploadResult, err := operation.UploadData(targetUrl, "", broker.option.Cipher, data, false, "", nil, assignResult.Auth)
  75. if err != nil {
  76. return nil, nil, fmt.Errorf("upload data %s: %v", targetUrl, err)
  77. }
  78. // println("uploaded to", targetUrl)
  79. return assignResult, uploadResult, nil
  80. }
  81. func (broker *MessageBroker) WithFilerClient(fn func(filer_pb.SeaweedFilerClient) error) (err error) {
  82. for _, filer := range broker.option.Filers {
  83. if err = pb.WithFilerClient(filer, broker.grpcDialOption, fn); err != nil {
  84. glog.V(0).Infof("fail to connect to %s: %v", filer, err)
  85. } else {
  86. break
  87. }
  88. }
  89. return
  90. }