GODRIVER-1648 Add check and test for closed cursor in Next (#428)

Author: KritiRav

mongo/change_stream.go

@@ -495,7 +495,7 @@ func (cs *ChangeStream) TryNext(ctx context.Context) bool {
 }
 
 func (cs *ChangeStream) next(ctx context.Context, nonBlocking bool) bool {
-	// return false right away if the change stream has already errored.
+	// return false right away if the change stream has already errored or if cursor is closed.
 	if cs.err != nil {
 		return false
 	}
@@ -538,6 +538,11 @@ func (cs *ChangeStream) loopNext(ctx context.Context, nonBlocking bool) {
 
 		cs.err = replaceErrors(cs.cursor.Err())
 		if cs.err == nil {
+			// Check if cursor is alive
+			if cs.ID() == 0 {
+				return
+			}
+
 			// If a getMore was done but the batch was empty, the batch cursor will return false with no error.
 			// Update the tracked resume token to catch the post batch resume token from the server response.
 			cs.updatePbrtFromCommand()

mongo/integration/change_stream_test.go

@@ -602,6 +602,28 @@ func TestChangeStream_ReplicaSet(t *testing.T) {
 			assert.Equal(mt, 2, numClearedEvents, "expected two PoolCleared events, got %d", numClearedEvents)
 		})
 	})
+	// Setting min server version as 4.0 since v3.6 does not send a "dropEvent"
+	mt.RunOpts("call to cursor.Next after cursor closed", mtest.NewOptions().MinServerVersion("4.0"), func(mt *mtest.T) {
+		cs, err := mt.Coll.Watch(mtest.Background, mongo.Pipeline{})
+		assert.Nil(mt, err, "Watch error: %v", err)
+		defer closeStream(cs)
+
+		// Generate insert events
+		generateEvents(mt, 5)
+		// Call Coll.Drop to generate drop and invalidate event
+		err = mt.Coll.Drop(mtest.Background)
+		assert.Nil(mt, err, "Drop error: %v", err)
+
+		// Test that all events were successful
+		for i := 0; i < 7; i++ {
+			assert.True(mt, cs.Next(mtest.Background), "Next returned false at index %d; iteration error: %v", i, cs.Err())
+		}
+
+		operationType := cs.Current.Lookup("operationType").StringValue()
+		assert.Equal(mt, operationType, "invalidate", "expected invalidate event but returned %q event", operationType)
+		// next call to cs.Next should return False since cursor is closed
+		assert.False(mt, cs.Next(mtest.Background), "expected to return false, but returned true")
+	})
 }
 
 func closeStream(cs *mongo.ChangeStream) {