diff --git a/codec_test.go b/codec_test.go index c84a871..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,10 +31,7 @@ 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() + ev, _, err := dec.Decode() if err != nil { t.Fatal(err) } diff --git a/decoder.go b/decoder.go index 1db90e7..e58a0b5 100644 --- a/decoder.go +++ b/decoder.go @@ -34,7 +34,9 @@ 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) { +// 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 _, err := dec.Peek(1) @@ -42,20 +44,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 +77,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..c2b37fc 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,28 +110,66 @@ 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 } - pub := ev.(*publication) - if pub.Retry() > 0 { - stream.retry = time.Duration(pub.Retry()) * time.Millisecond + + if comment != nil { + stream.Comments <- *comment } - if len(pub.Id()) > 0 { - stream.lastEventId = pub.Id() + + 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") } - stream.Events <- ev } - backoff := stream.retry for { + backoff := stream.backoffWithJitter(reconnectAttempts) + reconnectAttempts++ time.Sleep(backoff) if stream.Logger != nil { stream.Logger.Printf("Reconnecting in %0.4f secs\n", backoff.Seconds()) @@ -139,6 +184,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..89775b8 --- /dev/null +++ b/stream_test.go @@ -0,0 +1,43 @@ +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)) + } +} + +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)) + } + +}