mirror/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-11-27 20:59:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

filer_http: support uploading file with offset

Browse Source
This commit is contained in:
banjiaojuhao 2022-02-22 00:15:00 +08:00
parent 497ebbbd45
commit 6ab09e9071
2 changed files with 32 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
weed/server/filer_server_handlers_write_autochunk.go
View File

@ -126,10 +126,6 @@ func (fs *FilerServer) doPutAutoChunk(ctx context.Context, w http.ResponseWriter
return
}
func isAppend(r *http.Request) bool {
return r.URL.Query().Get("op") == "append"
}
func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileName string, contentType string, so *operation.StorageOption, md5bytes []byte, fileChunks []*filer_pb.FileChunk, chunkOffset int64, content []byte) (filerResult *FilerPostResult, replyerr error) {
// detect file mode
@ -161,8 +157,11 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa
var entry *filer.Entry
var mergedChunks []*filer_pb.FileChunk
isAppend := r.URL.Query().Get("op") == "append"
isOffsetWrite := fileChunks[0].Offset > 0
// when it is an append
if isAppend(r) {
if isAppend || isOffsetWrite {
existingEntry, findErr := fs.filer.FindEntry(ctx, util.FullPath(path))
if findErr != nil && findErr != filer_pb.ErrNotFound {
glog.V(0).Infof("failing to find %s: %v", path, findErr)
@ -173,11 +172,13 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa
entry.Mtime = time.Now()
entry.Md5 = nil
// adjust chunk offsets
for _, chunk := range fileChunks {
chunk.Offset += int64(entry.FileSize)
if isAppend {
for _, chunk := range fileChunks {
chunk.Offset += int64(entry.FileSize)
}
entry.FileSize += uint64(chunkOffset)
}
mergedChunks = append(entry.Chunks, fileChunks...)
entry.FileSize += uint64(chunkOffset)
// TODO
if len(entry.Content) > 0 {
@ -215,6 +216,10 @@ func (fs *FilerServer) saveMetaData(ctx context.Context, r *http.Request, fileNa
return
}
entry.Chunks = mergedChunks
if isOffsetWrite {
entry.Md5 = nil
entry.FileSize = entry.Size()
}
filerResult = &FilerPostResult{
Name: fileName,

20
weed/server/filer_server_handlers_write_upload.go
View File

@ -3,10 +3,12 @@ package weed_server
import (
"bytes"
"crypto/md5"
"fmt"
"hash"
"io"
"net/http"
"sort"
"strconv"
"strings"
"sync"
"sync/atomic"
@ -28,6 +30,22 @@ var bufPool = sync.Pool{
}
func (fs *FilerServer) uploadReaderToChunks(w http.ResponseWriter, r *http.Request, reader io.Reader, chunkSize int32, fileName, contentType string, contentLength int64, so *operation.StorageOption) (fileChunks []*filer_pb.FileChunk, md5Hash hash.Hash, chunkOffset int64, uploadErr error, smallContent []byte) {
query := r.URL.Query()
isAppend := query.Get("op") == "append"
if query.Has("offset") {
offset := query.Get("offset")
offsetInt, err := strconv.ParseInt(offset, 10, 64)
if err != nil || offsetInt < 0 {
err = fmt.Errorf("invalid 'offset': '%s'", offset)
return nil, nil, 0, err, nil
}
if isAppend && offsetInt > 0 {
err = fmt.Errorf("cannot set offset when op=append")
return nil, nil, 0, err, nil
}
chunkOffset = offsetInt
}
md5Hash = md5.New()
var partReader = io.NopCloser(io.TeeReader(reader, md5Hash))
@ -63,7 +81,7 @@ func (fs *FilerServer) uploadReaderToChunks(w http.ResponseWriter, r *http.Reque
bytesBufferLimitCond.Signal()
break
}
if chunkOffset == 0 && !isAppend(r) {
if chunkOffset == 0 && !isAppend {
if dataSize < fs.option.SaveToFilerLimit || strings.HasPrefix(r.URL.Path, filer.DirectoryEtcRoot) {
chunkOffset += dataSize
smallContent = make([]byte, dataSize)