🧱 Chunk is a download manager for slow and unstable servers
Merge pull request #9 from cuducos/test_improvements
Test and code improvements
authored by
Daniel Fireman
and committed by
GitHub
beb5eee5
37c0402a
+170
-82
+1
.gitignore
+1
.gitignore
···
1
+
chunk
+20
-29
main.go
+20
-29
main.go
···
7
7
"log"
8
8
"net/http"
9
9
"os"
10
+
"path/filepath"
10
11
"sync"
11
12
"time"
12
13
···
55
56
type Downloader struct {
56
57
// Client is the HTTP client used for every request needed to download all
57
58
// the files.
58
-
Client *http.Client
59
+
client *http.Client
59
60
60
61
// TimeoutPerChunk is the timeout for the download of each chunk from each
61
62
// URL. A chunk is a part of a file requested using the content range HTTP
···
90
91
if err != nil {
91
92
return nil, fmt.Errorf("error creating the request for %s: %w", u, err)
92
93
}
93
-
req = req.WithContext(ctx)
94
-
resp, err := d.Client.Do(req)
94
+
resp, err := d.client.Do(req)
95
95
if err != nil {
96
96
return nil, fmt.Errorf("error sending a get http request to %s: %w", u, err)
97
97
}
···
108
108
}
109
109
110
110
func (d *Downloader) downloadFileWithTimeout(userCtx context.Context, u string) ([]byte, error) {
111
-
ctx, cancel := context.WithTimeout(context.Background(), d.Client.Timeout)
111
+
ctx, cancel := context.WithTimeout(userCtx, d.TimeoutPerChunk) // need to propagate context, which might contain app-specific data.
112
112
defer cancel()
113
113
ch := make(chan []byte)
114
114
errs := make(chan error)
···
146
146
return nil
147
147
},
148
148
retry.Attempts(d.MaxRetriesPerChunk),
149
-
retry.MaxDelay(d.Client.Timeout),
149
+
retry.MaxDelay(d.WaitBetweenRetries),
150
150
)
151
151
if err != nil {
152
152
return nil, fmt.Errorf("error downloading %s: %w", u, err)
···
158
158
// DownloadWithContext is a version of Download that takes a context. The
159
159
// context can be used to stop all downloads in progress.
160
160
func (d *Downloader) DownloadWithContext(ctx context.Context, urls ...string) <-chan DownloadStatus {
161
+
if d.client == nil {
162
+
d.client = &http.Client{Timeout: d.TimeoutPerChunk}
163
+
}
161
164
ch := make(chan DownloadStatus)
162
165
var wg sync.WaitGroup
163
166
for _, u := range urls {
···
166
169
defer wg.Done()
167
170
s := DownloadStatus{URL: u}
168
171
defer func() { ch <- s }()
169
-
f, err := os.CreateTemp("", "chunk-download-")
170
-
if err != nil {
171
-
s.Error = err
172
-
return
173
-
}
174
-
s.DownloadedFilePath = f.Name()
172
+
s.DownloadedFilePath = filepath.Join(os.TempDir(), filepath.Base(u))
175
173
b, err := d.downloadFile(ctx, u)
176
174
if err != nil {
177
175
s.Error = err
178
176
return
179
177
}
180
-
if err := os.WriteFile(f.Name(), b, 0655); err != nil {
178
+
if err := os.WriteFile(s.DownloadedFilePath, b, 0655); err != nil {
181
179
s.Error = err
182
180
return
183
181
}
···
200
198
201
199
// NewDownloader creates a downloader with the defalt configuration. Check
202
200
// the constants in this package for their values.
203
-
func NewDownloader() *Downloader {
201
+
func DefaultDownloader() *Downloader {
204
202
return &Downloader{
205
-
&http.Client{Timeout: DefaultTimeoutPerChunk},
206
-
DefaultTimeoutPerChunk,
207
-
DefaultMaxParallelDownloadsPerServer,
208
-
DefaultMaxRetriesPerChunk,
209
-
DefaultChunkSize,
210
-
DefaultWaitBetweenRetries,
203
+
TimeoutPerChunk: DefaultTimeoutPerChunk,
204
+
MaxParallelDownloadsPerServer: DefaultMaxParallelDownloadsPerServer,
205
+
MaxRetriesPerChunk: DefaultMaxRetriesPerChunk,
206
+
ChunkSize: DefaultChunkSize,
207
+
WaitBetweenRetries: DefaultWaitBetweenRetries,
211
208
}
212
209
}
213
210
214
211
func main() {
215
-
d := NewDownloader()
216
-
for s := range d.Download(os.Args[1]) {
212
+
d := DefaultDownloader()
213
+
for s := range d.Download(os.Args[1:len(os.Args)]...) {
217
214
if s.Error != nil {
218
215
log.Fatal(s.Error)
219
216
}
220
217
if s.IsFinished() {
221
-
b, err := os.ReadFile(s.DownloadedFilePath)
222
-
if err != nil {
223
-
log.Fatal(err)
224
-
}
225
-
fmt.Print(string(b))
226
-
if err := os.Remove(s.DownloadedFilePath); err != nil {
227
-
log.Fatal(err)
228
-
}
218
+
log.Printf("Downloaded %s (%d bytes) to %s (%d bytes).", s.URL, s.FileSizeBytes, s.DownloadedFilePath, s.DownloadedFileBytes)
229
219
}
230
220
}
221
+
log.Println("All download(s) finished successfully.")
231
222
}
+149
-53
main_test.go
+149
-53
main_test.go
···
1
1
package main
2
2
3
3
import (
4
+
"context"
4
5
"fmt"
5
6
"net/http"
6
7
"net/http/httptest"
7
8
"os"
9
+
"strings"
8
10
"sync/atomic"
9
11
"testing"
10
12
"time"
11
13
)
12
14
13
-
func testServer(t *testing.T) *httptest.Server {
14
-
var attempt int32
15
-
paths := make(map[string]func(http.ResponseWriter))
16
-
paths["/ok"] = func(w http.ResponseWriter) {
17
-
fmt.Fprintf(w, "42")
15
+
func TestDownload_Error(t *testing.T) {
16
+
timeout := 250 * time.Millisecond
17
+
for _, tc := range []struct {
18
+
desc string
19
+
proc func(w http.ResponseWriter)
20
+
}{
21
+
{"failure", func(w http.ResponseWriter) { w.WriteHeader(http.StatusBadRequest) }},
22
+
{"timeout", func(w http.ResponseWriter) { time.Sleep(10 * timeout) }},
23
+
} {
24
+
t.Run(tc.desc, func(t *testing.T) {
25
+
s := httptest.NewServer(http.HandlerFunc(
26
+
func(w http.ResponseWriter, r *http.Request) {
27
+
tc.proc(w)
28
+
},
29
+
))
30
+
defer s.Close()
31
+
d := Downloader{
32
+
TimeoutPerChunk: timeout,
33
+
MaxRetriesPerChunk: 4,
34
+
MaxParallelDownloadsPerServer: 1,
35
+
ChunkSize: 1024,
36
+
WaitBetweenRetries: 0 * time.Second,
37
+
}
38
+
ch := d.Download(s.URL)
39
+
status := <-ch
40
+
if status.Error == nil {
41
+
t.Error("expected an error, but got nil")
42
+
}
43
+
if !strings.Contains(status.Error.Error(), "#4") {
44
+
t.Error("expected #4 (configured number of retries), but did not get it")
45
+
}
46
+
if _, ok := <-ch; ok {
47
+
t.Error("expected channel closed, but did not get it")
48
+
}
49
+
})
50
+
}
51
+
}
52
+
53
+
func TestDownload_OkWithDefaultDownloader(t *testing.T) {
54
+
s := httptest.NewServer(http.HandlerFunc(
55
+
func(w http.ResponseWriter, r *http.Request) {
56
+
fmt.Fprint(w, "42")
57
+
},
58
+
))
59
+
defer s.Close()
60
+
61
+
ch := DefaultDownloader().Download(s.URL)
62
+
got := <-ch
63
+
defer os.Remove(got.DownloadedFilePath)
64
+
65
+
if got.Error != nil {
66
+
t.Errorf("invalid error. want:nil got:%q", got.Error)
18
67
}
19
-
paths["/retry"] = func(w http.ResponseWriter) {
20
-
if atomic.LoadInt32(&attempt) == 0 {
21
-
atomic.StoreInt32(&attempt, 1)
22
-
time.Sleep(1 * time.Second)
23
-
}
24
-
fmt.Fprintf(w, "42")
68
+
if got.URL != s.URL {
69
+
t.Errorf("invalid URL. want:%s got:%s", s.URL, got.URL)
25
70
}
26
-
paths["/slow"] = func(w http.ResponseWriter) {
27
-
time.Sleep(1 * time.Second)
71
+
if got.DownloadedFileBytes != 2 {
72
+
t.Errorf("invalid DownloadedFileBytes. want:2 got:%d", got.DownloadedFileBytes)
28
73
}
29
-
30
-
return httptest.NewServer(
31
-
http.HandlerFunc(
32
-
func(w http.ResponseWriter, r *http.Request) {
33
-
h, ok := paths[r.URL.Path]
34
-
if !ok {
35
-
t.Fatalf("unknown url path for the test server %s", r.URL.Path)
36
-
}
37
-
h(w)
38
-
},
39
-
),
40
-
)
74
+
if got.FileSizeBytes != 2 {
75
+
t.Errorf("invalid FileSizeBytes. want:2 got:%d", got.FileSizeBytes)
76
+
}
77
+
b, err := os.ReadFile(got.DownloadedFilePath)
78
+
if err != nil {
79
+
t.Errorf("error reading downloaded file (%s): %q", got.DownloadedFilePath, err)
80
+
}
81
+
if string(b) != "42" {
82
+
t.Errorf("invalid downloaded file content. want:42 got:%s", string(b))
83
+
}
84
+
if _, ok := <-ch; ok {
85
+
t.Error("expected channel closed, but did not get it")
86
+
}
41
87
}
42
88
43
-
func TestDownload(t *testing.T) {
44
-
s := testServer(t)
45
-
defer s.Close()
89
+
func TestDownload_Retry(t *testing.T) {
90
+
timeout := 250 * time.Millisecond
46
91
for _, tc := range []struct {
47
-
desc string
48
-
path string
49
-
expected []byte
92
+
desc string
93
+
proc func(w http.ResponseWriter)
50
94
}{
51
-
{"normal response", "/ok", []byte("42")},
52
-
{"retried response", "/retry", []byte("42")},
53
-
{"timeout", "/slow", nil},
95
+
{"failure", func(w http.ResponseWriter) { w.WriteHeader(http.StatusBadRequest) }},
96
+
{"timeout", func(w http.ResponseWriter) { time.Sleep(10 * timeout) }},
54
97
} {
55
98
t.Run(tc.desc, func(t *testing.T) {
56
-
d := NewDownloader()
57
-
d.TimeoutPerChunk = 250 * time.Millisecond
58
-
d.Client.Timeout = 250 * time.Millisecond
59
-
d.MaxRetriesPerChunk = 3
60
-
ch := d.Download(s.URL + tc.path)
99
+
attempts := int32(0)
100
+
s := httptest.NewServer(http.HandlerFunc(
101
+
func(w http.ResponseWriter, r *http.Request) {
102
+
if atomic.CompareAndSwapInt32(&attempts, 0, 1) {
103
+
tc.proc(w)
104
+
}
105
+
fmt.Fprint(w, "42")
106
+
},
107
+
))
108
+
defer s.Close()
109
+
110
+
d := Downloader{
111
+
TimeoutPerChunk: timeout,
112
+
MaxRetriesPerChunk: 4,
113
+
MaxParallelDownloadsPerServer: 1,
114
+
ChunkSize: 1024,
115
+
WaitBetweenRetries: 0 * time.Second,
116
+
}
117
+
ch := d.Download(s.URL)
61
118
got := <-ch
62
-
var body []byte
63
-
if got.Error == nil {
64
-
var err error
65
-
body, err = os.ReadFile(got.DownloadedFilePath)
66
-
os.Remove(got.DownloadedFilePath)
67
-
if err != nil {
68
-
t.Errorf("could not read dowloaded file %s", got.DownloadedFilePath)
69
-
}
119
+
if got.Error != nil {
120
+
t.Errorf("invalid error. want:nil got:%q", got.Error)
121
+
}
122
+
if attempts != 1 {
123
+
t.Errorf("invalid number of attempts. want:1 got %d", attempts)
124
+
}
125
+
if got.URL != s.URL {
126
+
t.Errorf("invalid URL. want:%s got:%s", s.URL, got.URL)
127
+
}
128
+
if got.DownloadedFileBytes != 2 {
129
+
t.Errorf("invalid DownloadedFileBytes. want:2 got:%d", got.DownloadedFileBytes)
130
+
}
131
+
if got.FileSizeBytes != 2 {
132
+
t.Errorf("invalid FileSizeBytes. want:2 got:%d", got.FileSizeBytes)
70
133
}
71
-
if string(body) != string(tc.expected) {
72
-
t.Errorf("expected %s, got %s", string(tc.expected), string(body))
134
+
b, err := os.ReadFile(got.DownloadedFilePath)
135
+
if err != nil {
136
+
t.Errorf("error reading downloaded file (%s): %q", got.DownloadedFilePath, err)
73
137
}
74
-
if tc.expected == nil && got.Error == nil {
75
-
t.Error("expected an error, but got nil")
138
+
if string(b) != "42" {
139
+
t.Errorf("invalid downloaded file content. want:42 got:%s", string(b))
76
140
}
77
-
if tc.expected != nil && got.Error != nil {
78
-
t.Errorf("expected no error, but got %s", got.Error)
141
+
if _, ok := <-ch; ok {
142
+
t.Error("expected channel closed, but did not get it")
79
143
}
80
144
})
81
145
}
82
146
}
147
+
148
+
func TestDownloadWithContext_ErrorUserTimeout(t *testing.T) {
149
+
userTimeout := 250 * time.Millisecond // please note that the user timeout is less than the timeout per chunk.
150
+
timeout := 10 * userTimeout
151
+
s := httptest.NewServer(http.HandlerFunc(
152
+
func(w http.ResponseWriter, r *http.Request) {
153
+
time.Sleep(2 * userTimeout) // this time is greater than the user timeout, but shorter than the timeout per chunk.
154
+
},
155
+
))
156
+
defer s.Close()
157
+
d := Downloader{
158
+
TimeoutPerChunk: timeout,
159
+
MaxRetriesPerChunk: 4,
160
+
MaxParallelDownloadsPerServer: 1,
161
+
ChunkSize: 1024,
162
+
WaitBetweenRetries: 0 * time.Second,
163
+
}
164
+
userCtx, cancFunc := context.WithTimeout(context.Background(), userTimeout)
165
+
defer cancFunc()
166
+
167
+
ch := d.DownloadWithContext(userCtx, s.URL)
168
+
status := <-ch
169
+
if status.Error == nil {
170
+
t.Error("expected an error, but got nil")
171
+
}
172
+
if !strings.Contains(status.Error.Error(), "#4") {
173
+
t.Error("expected #4 (configured number of retries), but did not get it")
174
+
}
175
+
if _, ok := <-ch; ok {
176
+
t.Error("expected channel closed, but did not get it")
177
+
}
178
+
}