client: reconnect after an EOF

client.go

@@ -10,7 +10,6 @@ import (
 	"encoding/base64"
 	"errors"
 	"fmt"
-	"io"
 	"net/http"
 	"sync"
 	"sync/atomic"
@@ -213,10 +212,6 @@ func (c *Client) readLoop(reader *EventStreamReader, outCh chan *Event, erChan c
 		// Read each new line and process the type of event
 		event, err := reader.ReadEvent()
 		if err != nil {
-			if err == io.EOF {
-				erChan <- nil
-				return
-			}
 			// run user specified disconnect function
 			if c.disconnectcb != nil {
 				c.Connected = false

client_test.go

@@ -8,6 +8,7 @@ import (
 	"context"
 	"crypto/rand"
 	"encoding/hex"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"runtime"
@@ -258,6 +259,42 @@ func TestClientChanReconnect(t *testing.T) {
 	c.Unsubscribe(events)
 }
 
+func TestClientChanReconnectOnEOF(t *testing.T) {
+	setup(false)
+	defer cleanup()
+	streamID := "test"
+
+	c := NewClient(urlPath)
+
+	var (
+		reconnectErr      error
+		reconnectDuration time.Duration
+	)
+	c.ReconnectStrategy = backoff.NewConstantBackOff(time.Millisecond * 10)
+	c.ReconnectNotify = func(err error, duration time.Duration) {
+		reconnectErr = err
+		reconnectDuration = duration
+	}
+
+	events := make(chan *Event)
+	err := c.SubscribeChan(streamID, events)
+	require.Nil(t, err)
+
+	msg, err := wait(events, time.Millisecond*100)
+	assert.NoError(t, err)
+	assert.Equal(t, []byte(`ping`), msg)
+
+	srv.RemoveStream(streamID)
+	srv.CreateStream(streamID)
+	msg, err = wait(events, time.Millisecond*100)
+	assert.NoError(t, err)
+	assert.Equal(t, io.EOF, reconnectErr)
+	assert.Equal(t, time.Millisecond*10, reconnectDuration)
+	assert.Equal(t, []byte(`ping`), msg)
+
+	c.Unsubscribe(events)
+}
+
 func TestClientUnsubscribe(t *testing.T) {
 	setup(false)
 	defer cleanup()