mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/chrislusf/seaweedfs synced 2024-06-30 22:31:06 +02:00
Code Issues Projects Releases Wiki Activity

filer: retryable data chunk upload

Browse source
This commit is contained in:
chrislu 2022-08-20 19:15:44 -07:00
parent 4081d50607
commit f8fa430257
1 changed files with 30 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

48
weed/server/filer_server_handlers_write_autochunk.go
View file

@ -256,25 +256,37 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa
func (fs *FilerServer) saveAsChunk(so *operation.StorageOption) filer.SaveDataAsChunkFunctionType {
return func(reader io.Reader, name string, offset int64) (*filer_pb.FileChunk, error) {
// assign one file id for one chunk
fileId, urlLocation, auth, assignErr := fs.assignNewFileInfo(so)
if assignErr != nil {
return nil, assignErr
}
var fileId string
var uploadResult *operation.UploadResult
// upload the chunk to the volume server
uploadOption := &operation.UploadOption{
UploadUrl: urlLocation,
Filename: name,
Cipher: fs.option.Cipher,
IsInputCompressed: false,
MimeType: "",
PairMap: nil,
Jwt: auth,
}
uploadResult, uploadErr, _ := operation.Upload(reader, uploadOption)
if uploadErr != nil {
return nil, uploadErr
err := util.Retry("saveAsChunk", func() error {
// assign one file id for one chunk
assignedFileId, urlLocation, auth, assignErr := fs.assignNewFileInfo(so)
if assignErr != nil {
return assignErr
}
fileId = assignedFileId
// upload the chunk to the volume server
uploadOption := &operation.UploadOption{
UploadUrl: urlLocation,
Filename: name,
Cipher: fs.option.Cipher,
IsInputCompressed: false,
MimeType: "",
PairMap: nil,
Jwt: auth,
}
var uploadErr error
uploadResult, uploadErr, _ = operation.Upload(reader, uploadOption)
if uploadErr != nil {
return uploadErr
}
return nil
})
if err != nil {
return nil, err
}
return uploadResult.ToPbFileChunk(fileId, offset), nil