|
@ -44,8 +44,37 @@ func StreamContent(masterClient *wdclient.MasterClient, w io.Writer, chunks []*f |
|
|
|
|
|
|
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
// ---------------- ReadAllReader ----------------------------------
|
|
|
|
|
|
|
|
|
|
|
|
func ReadAll(masterClient *wdclient.MasterClient, chunks []*filer_pb.FileChunk) ([]byte, error) { |
|
|
|
|
|
|
|
|
|
|
|
buffer := bytes.Buffer{} |
|
|
|
|
|
|
|
|
|
|
|
chunkViews := ViewFromChunks(chunks, 0, math.MaxInt32) |
|
|
|
|
|
|
|
|
|
|
|
lookupFileId := func(fileId string) (targetUrl string, err error) { |
|
|
|
|
|
return masterClient.LookupFileId(fileId) |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
for _, chunkView := range chunkViews { |
|
|
|
|
|
urlString, err := lookupFileId(chunkView.FileId) |
|
|
|
|
|
if err != nil { |
|
|
|
|
|
glog.V(1).Infof("operation LookupFileId %s failed, err: %v", chunkView.FileId, err) |
|
|
|
|
|
return nil, err |
|
|
|
|
|
} |
|
|
|
|
|
err = util.ReadUrlAsStream(urlString, chunkView.CipherKey, chunkView.IsGzipped, chunkView.IsFullChunk(), chunkView.Offset, int(chunkView.Size), func(data []byte) { |
|
|
|
|
|
buffer.Write(data) |
|
|
|
|
|
}) |
|
|
|
|
|
if err != nil { |
|
|
|
|
|
glog.V(1).Infof("read %s failed, err: %v", chunkView.FileId, err) |
|
|
|
|
|
return nil, err |
|
|
|
|
|
} |
|
|
|
|
|
} |
|
|
|
|
|
return buffer.Bytes(), nil |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
// ---------------- ChunkStreamReader ----------------------------------
|
|
|
type ChunkStreamReader struct { |
|
|
type ChunkStreamReader struct { |
|
|
masterClient *wdclient.MasterClient |
|
|
|
|
|
chunkViews []*ChunkView |
|
|
chunkViews []*ChunkView |
|
|
logicOffset int64 |
|
|
logicOffset int64 |
|
|
buffer []byte |
|
|
buffer []byte |
|
@ -69,6 +98,7 @@ func NewChunkStreamReaderFromFiler(masterClient *wdclient.MasterClient, chunks [ |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
func (c *ChunkStreamReader) Read(p []byte) (n int, err error) { |
|
|
func (c *ChunkStreamReader) Read(p []byte) (n int, err error) { |
|
|
if c.isBufferEmpty() { |
|
|
if c.isBufferEmpty() { |
|
|
if c.chunkIndex >= len(c.chunkViews) { |
|
|
if c.chunkIndex >= len(c.chunkViews) { |
|
@ -144,6 +174,10 @@ func (c *ChunkStreamReader) fetchChunkToBuffer(chunkView *ChunkView) error { |
|
|
return nil |
|
|
return nil |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
func (c *ChunkStreamReader) Close() { |
|
|
|
|
|
// TODO try to release and reuse buffer
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
func VolumeId(fileId string) string { |
|
|
func VolumeId(fileId string) string { |
|
|
lastCommaIndex := strings.LastIndex(fileId, ",") |
|
|
lastCommaIndex := strings.LastIndex(fileId, ",") |
|
|
if lastCommaIndex > 0 { |
|
|
if lastCommaIndex > 0 { |
|
|