mirror of
https://github.com/rclone/rclone.git
synced 2026-11-07 10:12:56 -05:00
kevin9327
337c130c12
Seek documents that a position past the cached data returns the maximum usable offset with an error, but it returned offset - (abs - cacheLen), which is only the cache length for io.SeekStart. For io.SeekCurrent and io.SeekEnd it returned an arbitrary, possibly negative, number.
104 lines
3.0 KiB
Go
104 lines
3.0 KiB
Go
package readers
|
|
|
|
import (
|
|
"errors"
|
|
"io"
|
|
"sync"
|
|
)
|
|
|
|
// A RepeatableReader implements the io.ReadSeeker it allow to seek cached data
|
|
// back and forth within the reader but will only read data from the internal Reader as necessary
|
|
// and will play nicely with the Account and io.LimitedReader to reflect current speed
|
|
type RepeatableReader struct {
|
|
mu sync.Mutex // protect against concurrent use
|
|
in io.Reader // Input reader
|
|
i int64 // current reading index
|
|
b []byte // internal cache buffer
|
|
}
|
|
|
|
var _ io.ReadSeeker = (*RepeatableReader)(nil)
|
|
|
|
// Seek implements the io.Seeker interface.
|
|
// If seek position is passed the cache buffer length the function will return
|
|
// the maximum offset that can be used and "fs.RepeatableReader.Seek: offset is unavailable" Error
|
|
func (r *RepeatableReader) Seek(offset int64, whence int) (int64, error) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
|
|
var abs int64
|
|
cacheLen := int64(len(r.b))
|
|
switch whence {
|
|
case io.SeekStart:
|
|
abs = offset
|
|
case io.SeekCurrent:
|
|
abs = r.i + offset
|
|
case io.SeekEnd:
|
|
abs = cacheLen + offset
|
|
default:
|
|
return 0, errors.New("fs.RepeatableReader.Seek: invalid whence")
|
|
}
|
|
if abs < 0 {
|
|
return 0, errors.New("fs.RepeatableReader.Seek: negative position")
|
|
}
|
|
if abs > cacheLen {
|
|
return cacheLen, errors.New("fs.RepeatableReader.Seek: offset is unavailable")
|
|
}
|
|
r.i = abs
|
|
return abs, nil
|
|
}
|
|
|
|
// Read data from original Reader into bytes
|
|
// Data is either served from the underlying Reader or from cache if was already read
|
|
func (r *RepeatableReader) Read(b []byte) (n int, err error) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
|
|
cacheLen := int64(len(r.b))
|
|
if r.i == cacheLen {
|
|
n, err = r.in.Read(b)
|
|
if n > 0 {
|
|
r.b = append(r.b, b[:n]...)
|
|
}
|
|
} else {
|
|
n = copy(b, r.b[r.i:])
|
|
}
|
|
r.i += int64(n)
|
|
return n, err
|
|
}
|
|
|
|
// NewRepeatableReader create new repeatable reader from Reader r
|
|
func NewRepeatableReader(r io.Reader) *RepeatableReader {
|
|
return &RepeatableReader{in: r}
|
|
}
|
|
|
|
// NewRepeatableReaderSized create new repeatable reader from Reader r
|
|
// with an initial buffer of size.
|
|
func NewRepeatableReaderSized(r io.Reader, size int) *RepeatableReader {
|
|
return &RepeatableReader{
|
|
in: r,
|
|
b: make([]byte, 0, size),
|
|
}
|
|
}
|
|
|
|
// NewRepeatableLimitReader create new repeatable reader from Reader r
|
|
// with an initial buffer of size wrapped in an io.LimitReader to read
|
|
// only size.
|
|
func NewRepeatableLimitReader(r io.Reader, size int) *RepeatableReader {
|
|
return NewRepeatableReaderSized(io.LimitReader(r, int64(size)), size)
|
|
}
|
|
|
|
// NewRepeatableReaderBuffer create new repeatable reader from Reader r
|
|
// using the buffer passed in.
|
|
func NewRepeatableReaderBuffer(r io.Reader, buf []byte) *RepeatableReader {
|
|
return &RepeatableReader{
|
|
in: r,
|
|
b: buf[:0],
|
|
}
|
|
}
|
|
|
|
// NewRepeatableLimitReaderBuffer create new repeatable reader from
|
|
// Reader r and buf wrapped in an io.LimitReader to read only size.
|
|
func NewRepeatableLimitReaderBuffer(r io.Reader, buf []byte, size int64) *RepeatableReader {
|
|
return NewRepeatableReaderBuffer(io.LimitReader(r, size), buf)
|
|
}
|