mirror/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-01-18 06:30:07 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactoring

Browse Source
This commit is contained in:
Chris Lu 2020-10-08 23:19:42 -07:00
parent b2ee5873fb
commit 6e1f936efd
2 changed files with 18 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
weed/filer/filechunk_manifest.go
View File

@ -89,10 +89,15 @@ func fetchChunk(lookupFileIdFn LookupFileIdFunctionType, fileId string, cipherKe
glog.Errorf("operation LookupFileId %s failed, err: %v", fileId, err) glog.Errorf("operation LookupFileId %s failed, err: %v", fileId, err)
return nil, err return nil, err
} }
var buffer bytes.Buffer return fetchChunkData(urlStrings, cipherKey, isGzipped, true, 0, 0)
}
func fetchChunkData(urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, size int) ([]byte, error) {
var err error
var buffer bytes.Buffer
for _, urlString := range urlStrings { for _, urlString := range urlStrings {
err = util.ReadUrlAsStream(urlString, cipherKey, isGzipped, true, 0, 0, func(data []byte) { err = util.ReadUrlAsStream(urlString, cipherKey, isGzipped, isFullChunk, offset, size, func(data []byte) {
buffer.Write(data) buffer.Write(data)
}) })
if err != nil { if err != nil {

32
weed/filer/stream.go
View File

@ -32,18 +32,13 @@ func StreamContent(masterClient *wdclient.MasterClient, w io.Writer, chunks []*f
for _, chunkView := range chunkViews { for _, chunkView := range chunkViews {
urlStrings := fileId2Url[chunkView.FileId] urlStrings := fileId2Url[chunkView.FileId]
for _, urlString := range urlStrings {
err := util.ReadUrlAsStream(urlString, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.Offset, int(chunkView.Size), func(data []byte) { data, err := fetchChunkData(urlStrings, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.Offset, int(chunkView.Size))
w.Write(data) if err == nil {
}) return err
if err != nil {
// data already written to w would be wrong
// but usually there are nothing written if fails to read
glog.V(1).Infof("read %s failed, err: %v", chunkView.FileId, err)
} else {
break
}
} }
w.Write(data)
} }
return nil return nil
@ -68,17 +63,12 @@ func ReadAll(masterClient *wdclient.MasterClient, chunks []*filer_pb.FileChunk)
glog.V(1).Infof("operation LookupFileId %s failed, err: %v", chunkView.FileId, err) glog.V(1).Infof("operation LookupFileId %s failed, err: %v", chunkView.FileId, err)
return nil, err return nil, err
} }
for _, urlString := range urlStrings {
err = util.ReadUrlAsStream(urlString, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.Offset, int(chunkView.Size), func(data []byte) { data, err := fetchChunkData(urlStrings, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.Offset, int(chunkView.Size))
buffer.Write(data) if err != nil {
}) return nil, err
if err != nil {
glog.V(1).Infof("read %s failed, err: %v", chunkView.FileId, err)
buffer.Reset()
} else {
break
}
} }
buffer.Write(data)
} }
return buffer.Bytes(), nil return buffer.Bytes(), nil
} }