@@ -350,6 +350,49 @@ public void Execute_should_return_expected_results_for_inserts(
350350 }
351351 }
352352
353+ [ SkippableTheory ]
354+ [ ParameterAttributeData ]
355+ public void Execute_should_return_expected_results_for_large_batch (
356+ [ Values ( 1 , 2 , 3 ) ] int numberOfChunks ,
357+ [ Values ( false , true ) ] bool async )
358+ {
359+ RequireServer . Check ( ) . Supports ( Feature . ChangeStreamStage ) . ClusterTypes ( ClusterType . ReplicaSet , ClusterType . Sharded ) ;
360+ EnsureDatabaseExists ( ) ;
361+ DropCollection ( ) ;
362+
363+ var pipeline = new [ ] { BsonDocument . Parse ( "{ $match : { operationType : \" insert\" } }" ) } ;
364+ var resultSerializer = new ChangeStreamDocumentSerializer < BsonDocument > ( BsonDocumentSerializer . Instance ) ;
365+ var messageEncoderSettings = new MessageEncoderSettings ( ) ;
366+ var subject = new ChangeStreamOperation < ChangeStreamDocument < BsonDocument > > ( _collectionNamespace , pipeline , resultSerializer , messageEncoderSettings )
367+ {
368+ FullDocument = ChangeStreamFullDocumentOption . UpdateLookup
369+ } ;
370+ using ( var cursor = ExecuteOperation ( subject , async ) )
371+ {
372+ var filler = new string ( 'x' , ( numberOfChunks - 1 ) * 65536 ) ;
373+ var document = new BsonDocument { { "_id" , 1 } , { "filler" , filler } } ;
374+ Insert ( document ) ;
375+
376+ ChangeStreamDocument < BsonDocument > changeStreamDocument ;
377+ do
378+ {
379+ if ( async)
380+ {
381+ cursor . MoveNextAsync ( ) . GetAwaiter ( ) . GetResult ( ) ;
382+ }
383+ else
384+ {
385+ cursor . MoveNext ( ) ;
386+ }
387+
388+ changeStreamDocument = cursor . Current . FirstOrDefault ( ) ;
389+ }
390+ while ( changeStreamDocument == null ) ;
391+
392+ changeStreamDocument. FullDocument . Should ( ) . Be ( document ) ;
393+ }
394+ }
395+
353396 [ SkippableTheory ]
354397 [ ParameterAttributeData ]
355398 public void Execute_should_return_expected_results_for_updates(
0 commit comments