jstests/sharding/change_stream_transaction_sharded.js
/**
 * Asserts that the expected changes are found on the change stream cursor. Pushes the
 * corresponding change stream document (with resume token) to an array. When expected
 * changes are provided for both shards, we must assume that either shard's changes could
 * come first or that they are interleaved via applyOps index. This is because a cross shard
 * transaction may commit at a different cluster time on each shard, which impacts the
 * ordering of the change stream.
 */
Asserts that the expected changes are found on the change stream cursor. Pushes the
corresponding change stream document (with resume token) to an array. When expected
changes are provided for both shards, we must assume that either shard's changes could
come first or that they are interleaved via applyOps index. This is because a cross shard
transaction may commit at a different cluster time on each shard, which impacts the
ordering of the change stream.
Asserts that the expected changes are found on the change stream cursor . Pushes the corresponding change stream document ( with resume token ) to an array . When expected changes are provided for both shards we must assume that either shard ' s changes could come first or that they are interleaved via applyOps index . This is because a cross shard transaction may commit at a different cluster time on each shard which impacts the ordering of the change stream .
function assertWritesVisibleWithCapture(cursor,
                                        expectedChangesShard1,
                                        expectedChangesShard2,
                                        changeCaptureListShard1,
                                        changeCaptureListShard2) {
    function assertChangeEqualWithCapture(changeDoc, expectedChange, changeCaptureList) {
        assert.eq(expectedChange.operationType, changeDoc.operationType);
        assert.eq(expectedChange._id, changeDoc.documentKey._id);
        changeCaptureList.push(changeDoc);
    }
    while (expectedChangesShard1.length || expectedChangesShard2.length) {
        assert.soon(() => cursor.hasNext());
        const changeDoc = cursor.next();
        if (changeDoc.documentKey.shard === 1) {
            assert(expectedChangesShard1.length);
            assertChangeEqualWithCapture(
                changeDoc, expectedChangesShard1[0], changeCaptureListShard1);
            expectedChangesShard1.shift();
        } else {
            assert(changeDoc.documentKey.shard === 2);
            assert(expectedChangesShard2.length);
            assertChangeEqualWithCapture(
                changeDoc, expectedChangesShard2[0], changeCaptureListShard2);
            expectedChangesShard2.shift();
        }
    }
    assertNoChanges(cursor);
}
function assertWritesVisibleWithCapture ( cursor , expectedChangesShard1 , expectedChangesShard2 , changeCaptureListShard1 , changeCaptureListShard2 ) { function assertChangeEqualWithCapture ( changeDoc , expectedChange , changeCaptureList ) { assert . eq ( expectedChange . operationType , changeDoc . operationType ) ; assert . eq ( expectedChange . _id , changeDoc . documentKey . _id ) ; changeCaptureList . push ( changeDoc ) ; } while ( expectedChangesShard1 . length || expectedChangesShard2 . length ) { assert . soon ( ( ) => cursor . hasNext ( ) ) ; const changeDoc = cursor . next ( ) ; if ( changeDoc . documentKey . shard === 1 ) { assert ( expectedChangesShard1 . length ) ; assertChangeEqualWithCapture ( changeDoc , expectedChangesShard1 [ 0 ] , changeCaptureListShard1 ) ; expectedChangesShard1 . shift ( ) ; } else { assert ( changeDoc . documentKey . shard === 2 ) ; assert ( expectedChangesShard2 . length ) ; assertChangeEqualWithCapture ( changeDoc , expectedChangesShard2 [ 0 ] , changeCaptureListShard2 ) ; expectedChangesShard2 . shift ( ) ; } } assertNoChanges ( cursor ) ; }
Asserts that the expected changes are found on the change stream cursor.
Asserts that the expected changes are found on the change stream cursor .