Browse Source

Initial deliverer tests

Cory Slep 1 year ago
parent
commit
db08c01e85
2 changed files with 322 additions and 1 deletions
  1. 11 1
      deliverer/deliverer.go
  2. 311 0
      deliverer/deliverer_test.go

+ 11 - 1
deliverer/deliverer.go

@@ -125,7 +125,9 @@ func (r retryData) ShouldRetry(max int) bool {
125 125
 	return r.n < max
126 126
 }
127 127
 
128
-// Do spawns a goroutine that retries f until it returns no error.
128
+// Do spawns a goroutine that retries f until it returns no error. Retry
129
+// behavior is determined by the DeliveryOptions passed to the DelivererPool
130
+// upon construction.
129 131
 func (d *DelivererPool) Do(b []byte, to *url.URL, sendFn func([]byte, *url.URL) error) {
130 132
 	f := func() error {
131 133
 		return sendFn(b, to)
@@ -144,6 +146,10 @@ func (d *DelivererPool) Do(b []byte, to *url.URL, sendFn func([]byte, *url.URL)
144 146
 	}()
145 147
 }
146 148
 
149
+// Restart resumes a previous attempt at delivering a payload to the specified
150
+// URL. Retry behavior is determined by the DeliveryOptions passed to this
151
+// DelivererPool upon construction, and is not governed by the previous
152
+// DelivererPool that attempted to deliver the message.
147 153
 func (d *DelivererPool) Restart(b []byte, to *url.URL, id string, sendFn func([]byte, *url.URL) error) {
148 154
 	f := func() error {
149 155
 		return sendFn(b, to)
@@ -158,11 +164,14 @@ func (d *DelivererPool) Restart(b []byte, to *url.URL, id string, sendFn func([]
158 164
 	}()
159 165
 }
160 166
 
167
+// Stop turns down and stops any in-flight requests or retries.
161 168
 func (d *DelivererPool) Stop() {
162 169
 	d.cancel()
163 170
 	d.closeTimers()
164 171
 }
165 172
 
173
+// Provides a channel streaming any errors the pool encounters, including errors
174
+// that it retries on.
166 175
 func (d *DelivererPool) Errors() <-chan error {
167 176
 	return d.errChan
168 177
 }
@@ -188,6 +197,7 @@ func (d *DelivererPool) do(r retryData) {
188 197
 				d.persister.Undeliverable(r.id)
189 198
 			}
190 199
 		}
200
+		return
191 201
 	}
192 202
 	if d.persister != nil {
193 203
 		d.persister.Successful(r.id)

+ 311 - 0
deliverer/deliverer_test.go

@@ -0,0 +1,311 @@
1
+package deliverer
2
+
3
+import (
4
+	"fmt"
5
+	"github.com/go-test/deep"
6
+	"golang.org/x/time/rate"
7
+	"net/url"
8
+	"sync"
9
+	"testing"
10
+	"time"
11
+)
12
+
13
+const (
14
+	id1           = "id1"
15
+	id2           = "id2"
16
+	sending       = "sending"
17
+	cancel        = "cancel"
18
+	successful    = "successful"
19
+	retrying      = "retrying"
20
+	undeliverable = "undeliverable"
21
+	noState       = "noState"
22
+)
23
+
24
+var (
25
+	testBytes []byte = []byte{0, 1, 2, 3}
26
+	testURL   *url.URL
27
+)
28
+
29
+func init() {
30
+	var err error
31
+	testURL, err = url.Parse("example.com")
32
+	if err != nil {
33
+		panic(err)
34
+	}
35
+}
36
+
37
+var _ DeliveryPersister = &mockDeliveryPersister{}
38
+
39
+type mockDeliveryPersister struct {
40
+	t        *testing.T
41
+	i        int
42
+	mu       *sync.Mutex
43
+	id1State string
44
+	id2State string
45
+}
46
+
47
+func newMockDeliveryPersister(t *testing.T) *mockDeliveryPersister {
48
+	return &mockDeliveryPersister{
49
+		t:        t,
50
+		mu:       &sync.Mutex{},
51
+		id1State: noState,
52
+		id2State: noState,
53
+	}
54
+}
55
+
56
+func (m *mockDeliveryPersister) Sending(b []byte, to *url.URL) string {
57
+	m.mu.Lock()
58
+	defer m.mu.Unlock()
59
+	if m.i == 0 {
60
+		m.i++
61
+		return id1
62
+	} else if m.i == 1 {
63
+		m.i++
64
+		return id2
65
+	} else {
66
+		m.t.Fatal("too many calls to Sending")
67
+	}
68
+	return ""
69
+}
70
+
71
+func (m *mockDeliveryPersister) Cancel(id string) {
72
+	m.mu.Lock()
73
+	defer m.mu.Unlock()
74
+	if id == id1 {
75
+		m.id1State = cancel
76
+	} else if id == id2 {
77
+		m.id2State = cancel
78
+	} else {
79
+		m.t.Fatalf("unknown Cancel id: %s", id)
80
+	}
81
+}
82
+
83
+func (m *mockDeliveryPersister) Successful(id string) {
84
+	m.mu.Lock()
85
+	defer m.mu.Unlock()
86
+	if id == id1 {
87
+		m.id1State = successful
88
+	} else if id == id2 {
89
+		m.id2State = successful
90
+	} else {
91
+		m.t.Fatalf("unknown Successful id: %s", id)
92
+	}
93
+}
94
+
95
+func (m *mockDeliveryPersister) Retrying(id string) {
96
+	m.mu.Lock()
97
+	defer m.mu.Unlock()
98
+	if id == id1 {
99
+		m.id1State = retrying
100
+	} else if id == id2 {
101
+		m.id2State = retrying
102
+	} else {
103
+		m.t.Fatalf("unknown Retrying id: %s", id)
104
+	}
105
+}
106
+
107
+func (m *mockDeliveryPersister) Undeliverable(id string) {
108
+	m.mu.Lock()
109
+	defer m.mu.Unlock()
110
+	if id == id1 {
111
+		m.id1State = undeliverable
112
+	} else if id == id2 {
113
+		m.id2State = undeliverable
114
+	} else {
115
+		m.t.Fatalf("unknown Retrying id: %s", id)
116
+	}
117
+}
118
+
119
+func TestDelivererPoolSuccessNoPersister(t *testing.T) {
120
+	testSendFn := func(b []byte, u *url.URL) error {
121
+		if diff := deep.Equal(b, testBytes); diff != nil {
122
+			t.Fatal(diff)
123
+		} else if u != testURL {
124
+			t.Fatal("wrong testURL")
125
+		}
126
+		return nil
127
+	}
128
+	pool := NewDelivererPool(DeliveryOptions{
129
+		InitialRetryTime: time.Microsecond,
130
+		MaximumRetryTime: time.Microsecond,
131
+		BackoffFactor:    2,
132
+		MaxRetries:       1,
133
+		RateLimit:        rate.NewLimiter(1, 1),
134
+	})
135
+	pool.Do(testBytes, testURL, testSendFn)
136
+	time.Sleep(time.Microsecond * 500)
137
+}
138
+
139
+func TestDelivererPoolSuccessPersister(t *testing.T) {
140
+	testSendFn := func(b []byte, u *url.URL) error {
141
+		if diff := deep.Equal(b, testBytes); diff != nil {
142
+			t.Fatal(diff)
143
+		} else if u != testURL {
144
+			t.Fatal("wrong testURL")
145
+		}
146
+		return nil
147
+	}
148
+	p := newMockDeliveryPersister(t)
149
+	pool := NewDelivererPool(DeliveryOptions{
150
+		InitialRetryTime: time.Microsecond,
151
+		MaximumRetryTime: time.Microsecond,
152
+		BackoffFactor:    2,
153
+		MaxRetries:       1,
154
+		RateLimit:        rate.NewLimiter(1, 1),
155
+		Persister:        p,
156
+	})
157
+	pool.Do(testBytes, testURL, testSendFn)
158
+	time.Sleep(time.Microsecond * 500)
159
+	if p.id1State != successful {
160
+		t.Fatalf("want: %s, got %s", successful, p.id1State)
161
+	}
162
+}
163
+
164
+func TestRestartSuccess(t *testing.T) {
165
+	testSendFn := func(b []byte, u *url.URL) error {
166
+		if diff := deep.Equal(b, testBytes); diff != nil {
167
+			t.Fatal(diff)
168
+		} else if u != testURL {
169
+			t.Fatal("wrong testURL")
170
+		}
171
+		return nil
172
+	}
173
+	p := newMockDeliveryPersister(t)
174
+	pool := NewDelivererPool(DeliveryOptions{
175
+		InitialRetryTime: time.Microsecond,
176
+		MaximumRetryTime: time.Microsecond,
177
+		BackoffFactor:    2,
178
+		MaxRetries:       1,
179
+		RateLimit:        rate.NewLimiter(1, 1),
180
+		Persister:        p,
181
+	})
182
+	pool.Restart(testBytes, testURL, id2, testSendFn)
183
+	time.Sleep(time.Microsecond * 500)
184
+	if p.id2State != successful {
185
+		t.Fatalf("want: %s, got %s", successful, p.id1State)
186
+	}
187
+}
188
+
189
+func TestDelivererPoolRetrying(t *testing.T) {
190
+	testSendFn := func(b []byte, u *url.URL) error {
191
+		if diff := deep.Equal(b, testBytes); diff != nil {
192
+			t.Fatal(diff)
193
+		} else if u != testURL {
194
+			t.Fatal("wrong testURL")
195
+		}
196
+		return fmt.Errorf("expected")
197
+	}
198
+	p := newMockDeliveryPersister(t)
199
+	pool := NewDelivererPool(DeliveryOptions{
200
+		InitialRetryTime: time.Microsecond,
201
+		MaximumRetryTime: time.Microsecond,
202
+		BackoffFactor:    2,
203
+		MaxRetries:       1,
204
+		RateLimit:        rate.NewLimiter(1000000, 10000000),
205
+		Persister:        p,
206
+	})
207
+	pool.Do(testBytes, testURL, testSendFn)
208
+	time.Sleep(time.Microsecond * 500)
209
+	select {
210
+	case <-pool.Errors():
211
+	default:
212
+		t.Fatal("expected error")
213
+	}
214
+	time.Sleep(time.Microsecond * 500)
215
+	if p.id1State != retrying {
216
+		t.Fatalf("want: %s, got %s", retrying, p.id1State)
217
+	}
218
+}
219
+
220
+func TestDelivererPoolUndeliverable(t *testing.T) {
221
+	testSendFn := func(b []byte, u *url.URL) error {
222
+		if diff := deep.Equal(b, testBytes); diff != nil {
223
+			t.Fatal(diff)
224
+		} else if u != testURL {
225
+			t.Fatal("wrong testURL")
226
+		}
227
+		return fmt.Errorf("expected")
228
+	}
229
+	p := newMockDeliveryPersister(t)
230
+	pool := NewDelivererPool(DeliveryOptions{
231
+		InitialRetryTime: time.Microsecond,
232
+		MaximumRetryTime: time.Microsecond,
233
+		BackoffFactor:    2,
234
+		MaxRetries:       1,
235
+		RateLimit:        rate.NewLimiter(1000000, 10000000),
236
+		Persister:        p,
237
+	})
238
+	pool.Do(testBytes, testURL, testSendFn)
239
+	time.Sleep(time.Microsecond * 500)
240
+	<-pool.Errors()
241
+	time.Sleep(time.Microsecond * 500)
242
+	<-pool.Errors()
243
+	time.Sleep(time.Microsecond * 500)
244
+	<-pool.Errors()
245
+	time.Sleep(time.Microsecond * 500)
246
+	if p.id1State != undeliverable {
247
+		t.Fatalf("want: %s, got %s", undeliverable, p.id1State)
248
+	}
249
+}
250
+
251
+func TestRestartRetrying(t *testing.T) {
252
+	testSendFn := func(b []byte, u *url.URL) error {
253
+		if diff := deep.Equal(b, testBytes); diff != nil {
254
+			t.Fatal(diff)
255
+		} else if u != testURL {
256
+			t.Fatal("wrong testURL")
257
+		}
258
+		return fmt.Errorf("expected")
259
+	}
260
+	p := newMockDeliveryPersister(t)
261
+	pool := NewDelivererPool(DeliveryOptions{
262
+		InitialRetryTime: time.Microsecond,
263
+		MaximumRetryTime: time.Microsecond,
264
+		BackoffFactor:    2,
265
+		MaxRetries:       1,
266
+		RateLimit:        rate.NewLimiter(1000000, 10000000),
267
+		Persister:        p,
268
+	})
269
+	pool.Restart(testBytes, testURL, id2, testSendFn)
270
+	time.Sleep(time.Microsecond * 500)
271
+	select {
272
+	case <-pool.Errors():
273
+	default:
274
+		t.Fatal("expected error")
275
+	}
276
+	time.Sleep(time.Microsecond * 500)
277
+	if p.id2State != retrying {
278
+		t.Fatalf("want: %s, got %s", retrying, p.id2State)
279
+	}
280
+}
281
+
282
+func TestRestartUndeliverable(t *testing.T) {
283
+	testSendFn := func(b []byte, u *url.URL) error {
284
+		if diff := deep.Equal(b, testBytes); diff != nil {
285
+			t.Fatal(diff)
286
+		} else if u != testURL {
287
+			t.Fatal("wrong testURL")
288
+		}
289
+		return fmt.Errorf("expected")
290
+	}
291
+	p := newMockDeliveryPersister(t)
292
+	pool := NewDelivererPool(DeliveryOptions{
293
+		InitialRetryTime: time.Microsecond,
294
+		MaximumRetryTime: time.Microsecond,
295
+		BackoffFactor:    2,
296
+		MaxRetries:       1,
297
+		RateLimit:        rate.NewLimiter(1000000, 10000000),
298
+		Persister:        p,
299
+	})
300
+	pool.Restart(testBytes, testURL, id2, testSendFn)
301
+	time.Sleep(time.Microsecond * 500)
302
+	<-pool.Errors()
303
+	time.Sleep(time.Microsecond * 500)
304
+	<-pool.Errors()
305
+	time.Sleep(time.Microsecond * 500)
306
+	<-pool.Errors()
307
+	time.Sleep(time.Microsecond * 500)
308
+	if p.id2State != undeliverable {
309
+		t.Fatalf("want: %s, got %s", undeliverable, p.id2State)
310
+	}
311
+}