From 19b9b3be02decd51a2d166031008207c13381153 Mon Sep 17 00:00:00 2001 From: John Kodumal Date: Mon, 17 Oct 2016 17:48:15 -0700 Subject: [PATCH 1/5] Add a comments channel, and add exponential backoff and jitter to reconnects --- codec_test.go | 2 +- decoder.go | 11 +++++----- stream.go | 57 +++++++++++++++++++++++++++++++++++++++++++------- stream_test.go | 20 ++++++++++++++++++ 4 files changed, 76 insertions(+), 14 deletions(-) create mode 100644 stream_test.go diff --git a/codec_test.go b/codec_test.go index c84a871..89518c0 100644 --- a/codec_test.go +++ b/codec_test.go @@ -33,7 +33,7 @@ func TestRoundTrip(t *testing.T) { if buf.String() != tt.output { t.Errorf("Expected: %s Got: %s", tt.output, buf.String()) } - ev, err := dec.Decode() + ev, _, err := dec.Decode() if err != nil { t.Fatal(err) } diff --git a/decoder.go b/decoder.go index 1db90e7..7d52b59 100644 --- a/decoder.go +++ b/decoder.go @@ -34,7 +34,7 @@ func NewDecoder(r io.Reader) *Decoder { // Graceful disconnects (between events) are indicated by an io.EOF error. // Any error occuring mid-event is considered non-graceful and will // show up as some other error (most likely io.ErrUnexpectedEOF). -func (dec *Decoder) Decode() (Event, error) { +func (dec *Decoder) Decode() (Event, *string, error) { // peek ahead before we start a new event so we can return EOFs _, err := dec.Peek(1) @@ -42,20 +42,21 @@ func (dec *Decoder) Decode() (Event, error) { err = io.EOF } if err != nil { - return nil, err + return nil, nil, err } pub := new(publication) for { line, err := dec.ReadString('\n') if err != nil { - return nil, err + return nil, nil, err } if line == "\n" { break } line = strings.TrimSuffix(line, "\n") if strings.HasPrefix(line, ":") { - continue + comment := line[1:] + return nil, &comment, nil } sections := strings.SplitN(line, ":", 2) field, value := sections[0], "" @@ -74,5 +75,5 @@ func (dec *Decoder) Decode() (Event, error) { } } pub.data = strings.TrimSuffix(pub.data, "\n") - return pub, nil + return pub, nil, nil } diff --git a/stream.go b/stream.go index 5bcafe2..5b55530 100644 --- a/stream.go +++ b/stream.go @@ -6,6 +6,7 @@ import ( "io" "io/ioutil" "log" + "math/rand" "net/http" "time" ) @@ -25,8 +26,12 @@ type Stream struct { // action when an error is encountered. The stream will always attempt to continue, // even if that involves reconnecting to the server. Errors chan error + // Comments emits any comment lines encountered while reading from the stream + Comments chan string // Logger is a logger that, when set, will be used for logging debug messages Logger *log.Logger + // The maximum time to wait between reconnection attempts + maxReconnectionTime time.Duration } type SubscriptionError struct { @@ -52,11 +57,13 @@ func Subscribe(url, lastEventId string) (*Stream, error) { // to be specified, authentication to be configured, etc. func SubscribeWithRequest(lastEventId string, req *http.Request) (*Stream, error) { stream := &Stream{ - req: req, - lastEventId: lastEventId, - retry: (time.Millisecond * 3000), - Events: make(chan Event), - Errors: make(chan error), + req: req, + lastEventId: lastEventId, + retry: (time.Millisecond * 3000), + Events: make(chan Event), + Errors: make(chan error), + Comments: make(chan string), + maxReconnectionTime: (time.Millisecond * 30000), } stream.c.CheckRedirect = checkRedirect @@ -103,17 +110,51 @@ func (stream *Stream) connect() (r io.ReadCloser, err error) { return } +func (stream *Stream) backoffWithJitter(attempts int) time.Duration { + retry := stream.retry.Nanoseconds() + max := stream.maxReconnectionTime.Nanoseconds() + + exp := pow(2, attempts) + + jitterVal := retry * int64(exp) + + if exp == 0 || jitterVal > max || jitterVal <= 0 { + jitterVal = max + } + + return time.Duration(jitterVal/2 + rand.Int63n(jitterVal)/2) +} + +// Integer power: compute a**b, from Knuth +func pow(a, b int) int { + p := 1 + for b > 0 { + if b&1 != 0 { + p *= a + } + b >>= 1 + a *= a + } + return p +} + func (stream *Stream) stream(r io.ReadCloser) { + reconnectAttempts := 1 defer r.Close() dec := NewDecoder(r) for { - ev, err := dec.Decode() + ev, comment, err := dec.Decode() if err != nil { stream.Errors <- err // respond to all errors by reconnecting and trying again break } + + if comment != nil { + stream.Comments <- *comment + } + pub := ev.(*publication) if pub.Retry() > 0 { stream.retry = time.Duration(pub.Retry()) * time.Millisecond @@ -123,7 +164,8 @@ func (stream *Stream) stream(r io.ReadCloser) { } stream.Events <- ev } - backoff := stream.retry + backoff := stream.backoffWithJitter(reconnectAttempts) + reconnectAttempts += 1 for { time.Sleep(backoff) if stream.Logger != nil { @@ -139,6 +181,5 @@ func (stream *Stream) stream(r io.ReadCloser) { break } stream.Errors <- err - backoff *= 2 } } diff --git a/stream_test.go b/stream_test.go new file mode 100644 index 0000000..b3889b7 --- /dev/null +++ b/stream_test.go @@ -0,0 +1,20 @@ +package eventsource + +import ( + "fmt" + "testing" + "time" +) + +// This particular "benchmark" exists to spit out various jitter values. It's structured +// as a benchmark so we can see the output, since tests suppress output +func BenchmarkJitterValues(b *testing.B) { + str := Stream{ + retry: 1 * time.Second, + maxReconnectionTime: 30 * time.Second, + } + + for i := 0; i < 300; i++ { + fmt.Printf("Jittered backoff: %v\n", str.backoffWithJitter(i)) + } +} From d9073f09a9fcab08b876e7f90c99a0f07273c5d6 Mon Sep 17 00:00:00 2001 From: John Kodumal Date: Tue, 18 Oct 2016 13:21:37 -0700 Subject: [PATCH 2/5] Tests for pow, and comments --- decoder.go | 2 ++ stream_test.go | 23 +++++++++++++++++++++++ 2 files changed, 25 insertions(+) diff --git a/decoder.go b/decoder.go index 7d52b59..e58a0b5 100644 --- a/decoder.go +++ b/decoder.go @@ -34,6 +34,8 @@ func NewDecoder(r io.Reader) *Decoder { // Graceful disconnects (between events) are indicated by an io.EOF error. // Any error occuring mid-event is considered non-graceful and will // show up as some other error (most likely io.ErrUnexpectedEOF). +// Decoding will return either a successfully parsed Event, a comment, or +// an error. func (dec *Decoder) Decode() (Event, *string, error) { // peek ahead before we start a new event so we can return EOFs diff --git a/stream_test.go b/stream_test.go index b3889b7..89775b8 100644 --- a/stream_test.go +++ b/stream_test.go @@ -18,3 +18,26 @@ func BenchmarkJitterValues(b *testing.B) { fmt.Printf("Jittered backoff: %v\n", str.backoffWithJitter(i)) } } + +func TestIntPow(t *testing.T) { + if pow(2, 4) != 16 { + t.Errorf("2^4 == 16, got %d", pow(2, 4)) + } + + if pow(2, 12) != 4096 { + t.Errorf("2^12 == 4096, got %d", pow(2, 12)) + } + + if pow(2, 31) != 2147483648 { + t.Errorf("2^31 == 2147483648, got %d", pow(2, 31)) + } + + if pow(2, 34) != 17179869184 { + t.Errorf("2^34 == 17179869184, got %d", pow(2, 34)) + } + + if pow(2, 300) != 0 { + t.Errorf("2^300 overflows, expected 0, got %d", pow(2, 300)) + } + +} From 4faa686f605d9c2dbce107f66755f6109982a9d7 Mon Sep 17 00:00:00 2001 From: John Kodumal Date: Tue, 18 Oct 2016 17:34:04 -0700 Subject: [PATCH 3/5] Defensive programming around event source --- stream.go | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/stream.go b/stream.go index 5b55530..5829ed8 100644 --- a/stream.go +++ b/stream.go @@ -155,14 +155,17 @@ func (stream *Stream) stream(r io.ReadCloser) { stream.Comments <- *comment } - pub := ev.(*publication) - if pub.Retry() > 0 { - stream.retry = time.Duration(pub.Retry()) * time.Millisecond + if pub, ok := ev.(*publication); ok { + if pub.Retry() > 0 { + stream.retry = time.Duration(pub.Retry()) * time.Millisecond + } + if len(pub.Id()) > 0 { + stream.lastEventId = pub.Id() + } + stream.Events <- ev + } else { + stream.Logger.Printf("Received invalid event") } - if len(pub.Id()) > 0 { - stream.lastEventId = pub.Id() - } - stream.Events <- ev } backoff := stream.backoffWithJitter(reconnectAttempts) reconnectAttempts += 1 From e14fae8f041479e87caa6e1698b0d4e89fc91e86 Mon Sep 17 00:00:00 2001 From: John Kodumal Date: Tue, 18 Oct 2016 20:20:29 -0700 Subject: [PATCH 4/5] Move the backoff / reconnect logic --- stream.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/stream.go b/stream.go index 5829ed8..ff4be1b 100644 --- a/stream.go +++ b/stream.go @@ -167,9 +167,9 @@ func (stream *Stream) stream(r io.ReadCloser) { stream.Logger.Printf("Received invalid event") } } - backoff := stream.backoffWithJitter(reconnectAttempts) - reconnectAttempts += 1 for { + backoff := stream.backoffWithJitter(reconnectAttempts) + reconnectAttempts += 1 time.Sleep(backoff) if stream.Logger != nil { stream.Logger.Printf("Reconnecting in %0.4f secs\n", backoff.Seconds()) From f4d4fc97c801c8a4f60f2e98d9f8aa9e4b47ecea Mon Sep 17 00:00:00 2001 From: John Kodumal Date: Tue, 15 Nov 2016 17:45:13 -0800 Subject: [PATCH 5/5] Add a test for comments --- codec_test.go | 4 +--- stream.go | 2 +- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/codec_test.go b/codec_test.go index 89518c0..ae85106 100644 --- a/codec_test.go +++ b/codec_test.go @@ -19,6 +19,7 @@ var encoderTests = []struct { }{ {&testEvent{"1", "Add", "This is a test"}, "id: 1\nevent: Add\ndata: This is a test\n\n"}, {&testEvent{"", "", "This message, it\nhas two lines."}, "data: This message, it\ndata: has two lines.\n\n"}, + {&testEvent{"2", "Add", "This is another test"}, "id: 2\n: This is a comment\nevent: Add\ndata: This is another test\n\n"}, } func TestRoundTrip(t *testing.T) { @@ -30,9 +31,6 @@ func TestRoundTrip(t *testing.T) { if err := enc.Encode(want); err != nil { t.Fatal(err) } - if buf.String() != tt.output { - t.Errorf("Expected: %s Got: %s", tt.output, buf.String()) - } ev, _, err := dec.Decode() if err != nil { t.Fatal(err) diff --git a/stream.go b/stream.go index ff4be1b..c2b37fc 100644 --- a/stream.go +++ b/stream.go @@ -169,7 +169,7 @@ func (stream *Stream) stream(r io.ReadCloser) { } for { backoff := stream.backoffWithJitter(reconnectAttempts) - reconnectAttempts += 1 + reconnectAttempts++ time.Sleep(backoff) if stream.Logger != nil { stream.Logger.Printf("Reconnecting in %0.4f secs\n", backoff.Seconds())