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
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
|
package stackless
import (
"bytes"
"compress/flate"
"compress/gzip"
"fmt"
"io"
"testing"
"time"
)
func TestCompressFlateSerial(t *testing.T) {
t.Parallel()
if err := testCompressFlate(); err != nil {
t.Fatalf("unexpected error: %v", err)
}
}
func TestCompressFlateConcurrent(t *testing.T) {
t.Parallel()
if err := testConcurrent(testCompressFlate, 10); err != nil {
t.Fatalf("unexpected error: %v", err)
}
}
func testCompressFlate() error {
return testWriter(func(w io.Writer) Writer {
zw, err := flate.NewWriter(w, flate.DefaultCompression)
if err != nil {
panic(fmt.Sprintf("BUG: unexpected error: %v", err))
}
return zw
}, func(r io.Reader) io.Reader {
return flate.NewReader(r)
})
}
func TestCompressGzipSerial(t *testing.T) {
t.Parallel()
if err := testCompressGzip(); err != nil {
t.Fatalf("unexpected error: %v", err)
}
}
func TestCompressGzipConcurrent(t *testing.T) {
t.Parallel()
if err := testConcurrent(testCompressGzip, 10); err != nil {
t.Fatalf("unexpected error: %v", err)
}
}
func testCompressGzip() error {
return testWriter(func(w io.Writer) Writer {
return gzip.NewWriter(w)
}, func(r io.Reader) io.Reader {
zr, err := gzip.NewReader(r)
if err != nil {
panic(fmt.Sprintf("BUG: cannot create gzip reader: %v", err))
}
return zr
})
}
func testWriter(newWriter NewWriterFunc, newReader func(io.Reader) io.Reader) error {
dstW := &bytes.Buffer{}
w := NewWriter(dstW, newWriter)
for i := 0; i < 5; i++ {
if err := testWriterReuse(w, dstW, newReader); err != nil {
return fmt.Errorf("unexpected error when re-using writer on iteration %d: %w", i, err)
}
dstW = &bytes.Buffer{}
w.Reset(dstW)
}
return nil
}
func testWriterReuse(w Writer, r io.Reader, newReader func(io.Reader) io.Reader) error {
wantW := &bytes.Buffer{}
mw := io.MultiWriter(w, wantW)
for i := 0; i < 30; i++ {
fmt.Fprintf(mw, "foobar %d\n", i)
if i%13 == 0 {
if err := w.Flush(); err != nil {
return fmt.Errorf("error on flush: %w", err)
}
}
}
w.Close()
zr := newReader(r)
data, err := io.ReadAll(zr)
if err != nil {
return fmt.Errorf("unexpected error: %w, data=%q", err, data)
}
wantData := wantW.Bytes()
if !bytes.Equal(data, wantData) {
return fmt.Errorf("unexpected data: %q. Expecting %q", data, wantData)
}
return nil
}
func testConcurrent(testFunc func() error, concurrency int) error {
ch := make(chan error, concurrency)
for i := 0; i < concurrency; i++ {
go func() {
ch <- testFunc()
}()
}
for i := 0; i < concurrency; i++ {
select {
case err := <-ch:
if err != nil {
return fmt.Errorf("unexpected error on goroutine %d: %w", i, err)
}
case <-time.After(time.Second):
return fmt.Errorf("timeout on goroutine %d", i)
}
}
return nil
}
|
