extern/rclone
1
0
Fork 0
mirror of https://github.com/rclone/rclone.git synced 2024-12-23 15:38:57 +01:00
Code Issues Packages Projects Releases Wiki Activity

operations: make ReOpen and NewReOpen public for re-use elsewhere

Browse Source
This commit is contained in:
Nick Craig-Wood 2020-02-13 14:27:50 +00:00
parent cd3c699f28
commit 7f15cc9556
4 changed files with 12 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
fs/operations/multithread.go
View File

@ -71,7 +71,7 @@ func (mc *multiThreadCopyState) copyStream(ctx context.Context, stream int) (err
fs.Debugf(mc.src, "multi-thread copy: stream %d/%d (%d-%d) size %v starting", stream+1, mc.streams, start, end, fs.SizeSuffix(end-start)) fs.Debugf(mc.src, "multi-thread copy: stream %d/%d (%d-%d) size %v starting", stream+1, mc.streams, start, end, fs.SizeSuffix(end-start))
rc, err := newReOpen(ctx, mc.src, nil, &fs.RangeOption{Start: start, End: end - 1}, fs.Config.LowLevelRetries) rc, err := NewReOpen(ctx, mc.src, nil, &fs.RangeOption{Start: start, End: end - 1}, fs.Config.LowLevelRetries)
if err != nil { if err != nil {
return errors.Wrap(err, "multpart copy: failed to open source") return errors.Wrap(err, "multpart copy: failed to open source")
} }

2
fs/operations/operations.go
View File

@ -403,7 +403,7 @@ func Copy(ctx context.Context, f fs.Fs, dst fs.Object, remote string, src fs.Obj
} }
} else { } else {
var in0 io.ReadCloser var in0 io.ReadCloser
in0, err = newReOpen(ctx, src, hashOption, nil, fs.Config.LowLevelRetries) in0, err = NewReOpen(ctx, src, hashOption, nil, fs.Config.LowLevelRetries)
if err != nil { if err != nil {
err = errors.Wrap(err, "failed to open source object") err = errors.Wrap(err, "failed to open source object")
} else { } else {

16
fs/operations/reopen.go
View File

@ -10,8 +10,8 @@ import (
"github.com/rclone/rclone/fs/fserrors" "github.com/rclone/rclone/fs/fserrors"
) )
// reOpen is a wrapper for an object reader which reopens the stream on error // ReOpen is a wrapper for an object reader which reopens the stream on error
type reOpen struct { type ReOpen struct {
ctx context.Context ctx context.Context
mu sync.Mutex // mutex to protect the below mu sync.Mutex // mutex to protect the below
src fs.Object // object to open src fs.Object // object to open
@ -30,14 +30,14 @@ var (
errorTooManyTries = errors.New("failed to reopen: too many retries") errorTooManyTries = errors.New("failed to reopen: too many retries")
) )
// newReOpen makes a handle which will reopen itself and seek to where it was on errors // NewReOpen makes a handle which will reopen itself and seek to where it was on errors
// //
// If hashOption is set this will be applied when reading from the start // If hashOption is set this will be applied when reading from the start
// //
// If rangeOption is set then this will applied when reading from the // If rangeOption is set then this will applied when reading from the
// start, and updated on retries. // start, and updated on retries.
func newReOpen(ctx context.Context, src fs.Object, hashOption *fs.HashesOption, rangeOption *fs.RangeOption, maxTries int) (rc io.ReadCloser, err error) { func NewReOpen(ctx context.Context, src fs.Object, hashOption *fs.HashesOption, rangeOption *fs.RangeOption, maxTries int) (rc io.ReadCloser, err error) {
h := &reOpen{ h := &ReOpen{
ctx: ctx, ctx: ctx,
src: src, src: src,
hashOption: hashOption, hashOption: hashOption,
@ -56,7 +56,7 @@ func newReOpen(ctx context.Context, src fs.Object, hashOption *fs.HashesOption,
// open the underlying handle - call with lock held // open the underlying handle - call with lock held
// //
// we don't retry here as the Open() call will itself have low level retries // we don't retry here as the Open() call will itself have low level retries
func (h *reOpen) open() error { func (h *ReOpen) open() error {
var optsArray [2]fs.OpenOption var optsArray [2]fs.OpenOption
var opts = optsArray[:0] var opts = optsArray[:0]
if h.read == 0 { if h.read == 0 {
@ -93,7 +93,7 @@ func (h *reOpen) open() error {
} }
// Read bytes retrying as necessary // Read bytes retrying as necessary
func (h *reOpen) Read(p []byte) (n int, err error) { func (h *ReOpen) Read(p []byte) (n int, err error) {
h.mu.Lock() h.mu.Lock()
defer h.mu.Unlock() defer h.mu.Unlock()
if h.err != nil { if h.err != nil {
@ -119,7 +119,7 @@ func (h *reOpen) Read(p []byte) (n int, err error) {
} }
// Close the stream // Close the stream
func (h *reOpen) Close() error { func (h *ReOpen) Close() error {
h.mu.Lock() h.mu.Lock()
defer h.mu.Unlock() defer h.mu.Unlock()
if !h.opened { if !h.opened {

4
fs/operations/reopen_test.go
View File

@ -15,7 +15,7 @@ import (
) )
// check interface // check interface
var _ io.ReadCloser = (*reOpen)(nil) var _ io.ReadCloser = (*ReOpen)(nil)
var errorTestError = errors.New("test error") var errorTestError = errors.New("test error")
@ -74,7 +74,7 @@ func TestReOpen(t *testing.T) {
breaks: breaks, breaks: breaks,
} }
hashOption := &fs.HashesOption{Hashes: hash.NewHashSet(hash.MD5)} hashOption := &fs.HashesOption{Hashes: hash.NewHashSet(hash.MD5)}
return newReOpen(context.Background(), src, hashOption, rangeOption, maxRetries) return NewReOpen(context.Background(), src, hashOption, rangeOption, maxRetries)
} }
t.Run("Basics", func(t *testing.T) { t.Run("Basics", func(t *testing.T) {