blob: 19dd1b2eae4e046771926564dfc3024b3ae2f4bf (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
|
package zw
import (
"context"
"io"
"io/fs"
"sync"
"sync/atomic"
)
type downloadReader struct {
recv func() ([]byte, error)
cancel context.CancelFunc
pending []byte
terminal error
cancelOnce sync.Once
closed atomic.Bool
}
func (r *downloadReader) Read(destination []byte) (int, error) {
if len(destination) == 0 {
return 0, nil
}
if r.closed.Load() {
return 0, fs.ErrClosed
}
if len(r.pending) > 0 {
n := copy(destination, r.pending)
r.pending = r.pending[n:]
return n, nil
}
if r.terminal != nil {
return 0, r.terminal
}
for {
chunk, err := r.recv()
if err != nil {
if r.closed.Load() {
return 0, fs.ErrClosed
}
if err == io.EOF {
r.terminal = io.EOF
} else {
r.terminal = convertRPCError(err)
}
r.cancelOnce.Do(r.cancel)
return 0, r.terminal
}
if len(chunk) == 0 {
continue
}
n := copy(destination, chunk)
r.pending = chunk[n:]
return n, nil
}
}
func (r *downloadReader) Close() error {
r.closed.Store(true)
r.cancelOnce.Do(r.cancel)
return nil
}
|
