seaweedfs/weed/messaging/broker/broker_append.go

131 lines
3.5 KiB
Go
Raw Normal View History

2020-04-18 08:12:01 +00:00
package broker
2020-04-17 09:29:38 +00:00
import (
"context"
"fmt"
"github.com/chrislusf/seaweedfs/weed/security"
2020-05-08 09:47:22 +00:00
"io"
2020-04-17 09:29:38 +00:00
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/operation"
"github.com/chrislusf/seaweedfs/weed/pb"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/pb/messaging_pb"
"github.com/chrislusf/seaweedfs/weed/util"
)
func (broker *MessageBroker) appendToFile(targetFile string, topicConfig *messaging_pb.TopicConfiguration, data []byte) error {
assignResult, uploadResult, err2 := broker.assignAndUpload(topicConfig, data)
if err2 != nil {
return err2
}
dir, name := util.FullPath(targetFile).DirAndName()
// append the chunk
if err := broker.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
request := &filer_pb.AppendToEntryRequest{
Directory: dir,
EntryName: name,
Chunks: []*filer_pb.FileChunk{uploadResult.ToPbFileChunk(assignResult.Fid, 0)},
2020-04-17 09:29:38 +00:00
}
_, err := client.AppendToEntry(context.Background(), request)
if err != nil {
glog.V(0).Infof("append to file %v: %v", request, err)
return err
}
return nil
}); err != nil {
return fmt.Errorf("append to file %v: %v", targetFile, err)
}
return nil
}
func (broker *MessageBroker) assignAndUpload(topicConfig *messaging_pb.TopicConfiguration, data []byte) (*operation.AssignResult, *operation.UploadResult, error) {
var assignResult = &operation.AssignResult{}
// assign a volume location
if err := broker.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
assignErr := util.Retry("assignVolume", func() error {
request := &filer_pb.AssignVolumeRequest{
Count: 1,
Replication: topicConfig.Replication,
Collection: topicConfig.Collection,
}
2020-04-17 09:29:38 +00:00
resp, err := client.AssignVolume(context.Background(), request)
if err != nil {
glog.V(0).Infof("assign volume failure %v: %v", request, err)
return err
}
if resp.Error != "" {
return fmt.Errorf("assign volume failure %v: %v", request, resp.Error)
}
assignResult.Auth = security.EncodedJwt(resp.Auth)
assignResult.Fid = resp.FileId
assignResult.Url = resp.Location.Url
assignResult.PublicUrl = resp.Location.PublicUrl
assignResult.GrpcPort = int(resp.Location.GrpcPort)
assignResult.Count = uint64(resp.Count)
2020-04-17 09:29:38 +00:00
return nil
})
if assignErr != nil {
return assignErr
}
2020-04-17 09:29:38 +00:00
return nil
}); err != nil {
return nil, nil, err
}
// upload data
targetUrl := fmt.Sprintf("http://%s/%s", assignResult.Url, assignResult.Fid)
2021-09-06 23:20:49 +00:00
uploadOption := &operation.UploadOption{
UploadUrl: targetUrl,
Filename: "",
Cipher: broker.option.Cipher,
IsInputCompressed: false,
MimeType: "",
PairMap: nil,
Jwt: assignResult.Auth,
}
uploadResult, err := operation.UploadData(data, uploadOption)
2020-04-17 09:29:38 +00:00
if err != nil {
return nil, nil, fmt.Errorf("upload data %s: %v", targetUrl, err)
}
// println("uploaded to", targetUrl)
return assignResult, uploadResult, nil
}
2020-04-29 20:26:02 +00:00
var _ = filer_pb.FilerClient(&MessageBroker{})
2020-04-17 09:29:38 +00:00
func (broker *MessageBroker) WithFilerClient(fn func(filer_pb.SeaweedFilerClient) error) (err error) {
for _, filer := range broker.option.Filers {
if err = pb.WithFilerClient(filer, broker.grpcDialOption, fn); err != nil {
2020-05-08 09:47:22 +00:00
if err == io.EOF {
return
}
2020-04-17 09:29:38 +00:00
glog.V(0).Infof("fail to connect to %s: %v", filer, err)
} else {
break
}
}
return
}
2021-01-28 22:36:29 +00:00
func (broker *MessageBroker) AdjustedUrl(location *filer_pb.Location) string {
return location.Url
}