Add a comments channel, and add exponential backoff and jitter to reconnects

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)
 		}

decoder.go

@@ -34,28 +34,31 @@ 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)
 	if err == io.ErrUnexpectedEOF {
 		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
 }

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
 	}
 }

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))
+	}
+
+}