GODRIVER-2992 [master] Add RemainingBatchLength to ChangeStream (#1633)

Co-authored-by: Felipe Gasper <[email protected]>

Author: blink1073

internal/integration/change_stream_test.go

@@ -98,6 +98,7 @@ func TestChangeStream_ReplicaSet(t *testing.T) {
 		// cause an event to occur so the resume token is updated
 		generateEvents(mt, 1)
 		assert.True(mt, cs.Next(context.Background()), "expected next to return true, got false")
+		assert.Equal(mt, 0, cs.RemainingBatchLength())
 		firstToken := cs.ResumeToken()
 
 		// cause an event on a different collection than the one being watched so the server's PBRT is updated
@@ -373,6 +374,7 @@ func TestChangeStream_ReplicaSet(t *testing.T) {
 
 						// Iterate over one event to get resume token
 						assert.True(mt, cs.Next(context.Background()), "expected Next to return true, got false")
+						assert.Equal(mt, numEvents-1, cs.RemainingBatchLength())
 						token := cs.ResumeToken()
 						closeStream(cs)
 

mongo/change_stream.go

@@ -579,6 +579,12 @@ func (cs *ChangeStream) ID() int64 {
 	return cs.cursor.ID()
 }
 
+// RemainingBatchLength returns the number of documents left in the current batch. If this returns zero, the subsequent
+// call to Next or TryNext will do a network request to fetch the next batch.
+func (cs *ChangeStream) RemainingBatchLength() int {
+	return len(cs.batch)
+}
+
 // SetBatchSize sets the number of documents to fetch from the database with
 // each iteration of the ChangeStream's "Next" or "TryNext" method. This setting
 // only affects subsequent document batches fetched from the database.