package chunkedreader import ( "io" "sync" "github.com/ncw/rclone/fs" ) // ChunkedReader is a reader for a Object with the possibility // of reading the source in chunks of given size // // A initialChunkSize of 0 will disable chunked reading. type ChunkedReader struct { mu sync.Mutex o fs.Object rc io.ReadCloser offset int64 chunkOffset int64 chunkSize int64 initialChunkSize int64 chunkGrowth bool doSeek bool } // New returns a ChunkedReader for the Object. // // A initialChunkSize of 0 will disable chunked reading. // If chunkGrowth is true, the chunk size will be doubled after each chunk read. // A Seek or RangeSeek will reset the chunk size to it's initial value func New(o fs.Object, initialChunkSize int64, chunkGrowth bool) *ChunkedReader { if initialChunkSize < 0 { initialChunkSize = 0 } return &ChunkedReader{ o: o, offset: -1, chunkSize: initialChunkSize, initialChunkSize: initialChunkSize, chunkGrowth: chunkGrowth, } } // Read from the file - for details see io.Reader func (cr *ChunkedReader) Read(p []byte) (n int, err error) { cr.mu.Lock() defer cr.mu.Unlock() for reqSize := int64(len(p)); reqSize > 0; reqSize = int64(len(p)) { chunkEnd := cr.chunkOffset + cr.chunkSize fs.Debugf(cr.o, "ChunkedReader.Read at %d length %d chunkOffset %d chunkSize %d", cr.offset, reqSize, cr.chunkOffset, cr.chunkSize) if atChunkEnd := cr.offset == chunkEnd; cr.offset == -1 || atChunkEnd { if atChunkEnd && cr.chunkSize > 0 { if cr.doSeek { cr.doSeek = false cr.chunkSize = cr.initialChunkSize } else if cr.chunkGrowth { cr.chunkSize *= 2 } cr.chunkOffset = cr.offset } err = cr.openRange() if err != nil { return } } var buf []byte chunkRest := chunkEnd - cr.offset if reqSize > chunkRest && cr.chunkSize != 0 { buf, p = p[0:chunkRest], p[chunkRest:] } else { buf, p = p, nil } var rn int rn, err = io.ReadFull(cr.rc, buf) n += rn cr.offset += int64(rn) if err != nil { return } } return n, nil } // Close the file - for details see io.Closer func (cr *ChunkedReader) Close() error { cr.mu.Lock() defer cr.mu.Unlock() return cr.resetReader(nil, 0) } // Seek the file - for details see io.Seeker func (cr *ChunkedReader) Seek(offset int64, whence int) (int64, error) { return cr.RangeSeek(offset, whence, -1) } // RangeSeek the file - for details see RangeSeeker func (cr *ChunkedReader) RangeSeek(offset int64, whence int, length int64) (int64, error) { cr.mu.Lock() defer cr.mu.Unlock() fs.Debugf(cr.o, "ChunkedReader.RangeSeek from %d to %d", cr.offset, offset) size := cr.o.Size() switch whence { case io.SeekStart: cr.offset = 0 case io.SeekEnd: cr.offset = size } cr.chunkOffset = cr.offset + offset cr.offset = -1 cr.doSeek = true if length > 0 { cr.chunkSize = length } else { cr.chunkSize = cr.initialChunkSize } return cr.offset, nil } // Open forces the connection to be opened func (cr *ChunkedReader) Open() (*ChunkedReader, error) { cr.mu.Lock() defer cr.mu.Unlock() return cr, cr.openRange() } // openRange will open the source Object with the given range // A length <= 0 will request till the end of the file func (cr *ChunkedReader) openRange() error { offset, length := cr.chunkOffset, cr.chunkSize fs.Debugf(cr.o, "ChunkedReader.openRange at %d length %d", offset, length) var rc io.ReadCloser var err error if length <= 0 { if offset == 0 { rc, err = cr.o.Open() } else { rc, err = cr.o.Open(&fs.RangeOption{Start: offset, End: -1}) } } else { rc, err = cr.o.Open(&fs.RangeOption{Start: offset, End: offset + length - 1}) } if err != nil { return err } return cr.resetReader(rc, offset) } // resetReader switches the current reader to the given reader. // The old reader will be Close'd before setting the new reader. func (cr *ChunkedReader) resetReader(rc io.ReadCloser, offset int64) error { if cr.rc != nil { if err := cr.rc.Close(); err != nil { return err } } cr.rc = rc cr.offset = offset return nil } var ( _ io.ReadCloser = (*ChunkedReader)(nil) _ io.Seeker = (*ChunkedReader)(nil) _ fs.RangeSeeker = (*ChunkedReader)(nil) )