Add nil checks for change stream cursors.

GODRIVER-464

Change-Id: I62b06a9fd52c9cd78d2fd32648d795705a98a102

Author: Divjot Arora

mongo/change_stream.go

@@ -31,6 +31,9 @@ const errorCursorKilled int32 = 237
 // contain a resume token.
 var ErrMissingResumeToken = errors.New("cannot provide resume functionality when the resume token is missing")
 
+// ErrNilCursor indicates that the cursor for the change stream is nil.
+var ErrNilCursor = errors.New("cursor is nil")
+
 type changeStream struct {
 	cmd      bsonx.Doc // aggregate command to run to create stream and rebuild cursor
 	pipeline bsonx.Arr
@@ -396,10 +399,18 @@ func newClientChangeStream(ctx context.Context, client *Client, pipeline interfa
 }
 
 func (cs *changeStream) ID() int64 {
+	if cs.cursor == nil {
+		return 0
+	}
+
 	return cs.cursor.ID()
 }
 
 func (cs *changeStream) Next(ctx context.Context) bool {
+	if cs.cursor == nil {
+		return false
+	}
+
 	if cs.cursor.Next(ctx) {
 		return true
 	}
@@ -431,6 +442,10 @@ func (cs *changeStream) Next(ctx context.Context) bool {
 }
 
 func (cs *changeStream) Decode(out interface{}) error {
+	if cs.cursor == nil {
+		return ErrNilCursor
+	}
+
 	br, err := cs.DecodeBytes()
 	if err != nil {
 		return err
@@ -440,6 +455,10 @@ func (cs *changeStream) Decode(out interface{}) error {
 }
 
 func (cs *changeStream) DecodeBytes() (bson.Raw, error) {
+	if cs.cursor == nil {
+		return nil, ErrNilCursor
+	}
+
 	br, err := cs.cursor.DecodeBytes()
 	if err != nil {
 		return nil, err
@@ -471,11 +490,18 @@ func (cs *changeStream) Err() error {
 	if cs.err != nil {
 		return cs.err
 	}
+	if cs.cursor == nil {
+		return nil
+	}
 
 	return cs.cursor.Err()
 }
 
 func (cs *changeStream) Close(ctx context.Context) error {
+	if cs.cursor == nil {
+		return nil // cursor is already closed
+	}
+
 	return cs.cursor.Close(ctx)
 }
 

mongo/change_stream_test.go

@@ -239,6 +239,29 @@ func TestChangeStream(t *testing.T) {
 			t.Errorf("Should have returned command error, but got %T", err)
 		}
 	})
+
+	t.Run("TestNilCursor", func(t *testing.T) {
+		cs := &changeStream{}
+
+		if id := cs.ID(); id != 0 {
+			t.Fatalf("Wrong ID returned. Expected 0 got %d", id)
+		}
+		if cs.Next(ctx) {
+			t.Fatalf("Next returned true, expected false")
+		}
+		if err := cs.Decode(nil); err != ErrNilCursor {
+			t.Fatalf("Wrong decode err. Expected ErrNilCursor got %s", err)
+		}
+		if _, err := cs.DecodeBytes(); err != ErrNilCursor {
+			t.Fatalf("Wrong decode bytes err. Expected ErrNilCursor got %s", err)
+		}
+		if err := cs.Err(); err != nil {
+			t.Fatalf("Wrong Err error. Expected nil got %s", err)
+		}
+		if err := cs.Close(ctx); err != nil {
+			t.Fatalf("Wrong Close error. Expected nil got %s", err)
+		}
+	})
 }
 
 func TestChangeStream_ReplicaSet(t *testing.T) {