fix(NODE-3648): hook operation into cursor

Author: durran

src/cursor/abstract_cursor.ts

@@ -14,7 +14,8 @@ import { ReadPreference, ReadPreferenceLike } from '../read_preference';
 import type { Server } from '../sdam/server';
 import type { Topology } from '../sdam/topology';
 import { Readable, Transform } from 'stream';
-import type { ExecutionResult } from '../operations/execute_operation';
+import { executeOperation, ExecutionResult } from '../operations/execute_operation';
+import { GetMoreOperation } from '../operations/get_more';
 import { ReadConcern, ReadConcernLike } from '../read_concern';
 import { TODO_NODE_3286, TypedEventEmitter } from '../mongo_types';
 
@@ -610,16 +611,13 @@ export abstract class AbstractCursor<
       return;
     }
 
-    server.getMore(
-      cursorNs,
-      cursorId,
-      {
-        ...this[kOptions],
-        session: this[kSession],
-        batchSize
-      },
-      callback
-    );
+    const getMoreOperation = new GetMoreOperation(cursorNs, cursorId, server, {
+      ...this[kOptions],
+      session: this[kSession],
+      batchSize
+    });
+
+    executeOperation(this.topology, getMoreOperation, callback);
   }
 }
 

src/operations/get_more.ts

@@ -37,16 +37,8 @@ export class GetMoreOperation extends AbstractOperation {
    * for execute passes a server so we will just use that one.
    */
   execute(server: Server, session: ClientSession, callback: Callback<Document>): void {
-    server.getMore(
-      this.ns,
-      this.cursorId,
-      {
-        ...this.options,
-        session: session
-      },
-      callback
-    );
+    server.getMore(this.ns, this.cursorId, this.options, callback);
   }
 }
 
-defineAspects(GetMoreOperation, [Aspect.READ_OPERATION, Aspect.RETRYABLE, Aspect.CURSOR_ITERATING]);
+defineAspects(GetMoreOperation, [Aspect.READ_OPERATION, Aspect.CURSOR_ITERATING]);

test/functional/change_stream_spec.test.js

@@ -65,6 +65,7 @@ describe('Change Stream Spec - v1', function () {
             ctx.database = ctx.client.db(sDB);
             ctx.collection = ctx.database.collection(sColl);
             ctx.client.on('commandStarted', e => {
+              console.log(e);
               if (e.commandName !== 'ismaster') _events.push(e);
             });
           });
@@ -170,6 +171,7 @@ describe('Change Stream Spec - v1', function () {
     const expectedEvents = test.expectations || [];
 
     return function testAPM(ctx, events) {
+      console.log('events', events);
       expectedEvents
         .map(e => e.command_started_event)
         .map(normalizeAPMEvent)

test/unit/operations/get_more.test.js

@@ -6,11 +6,17 @@ const { Long } = require('../../../src/bson');
 const { GetMoreOperation } = require('../../../src/operations/get_more');
 const { Server } = require('../../../src/sdam/server');
 const { ClientSession } = require('../../../src/sessions');
+const { ReadPreference } = require('../../../src/read_preference');
 
 describe('GetMoreOperation', function () {
   const ns = 'db.coll';
   const cursorId = Long.fromNumber(1);
-  const options = { batchSize: 100, comment: 'test', maxTimeMS: 500 };
+  const options = {
+    batchSize: 100,
+    comment: 'test',
+    maxTimeMS: 500,
+    readPreference: ReadPreference.primary
+  };
 
   describe('#constructor', function () {
     const server = sinon.createStubInstance(Server, {});
@@ -39,15 +45,16 @@ describe('GetMoreOperation', function () {
       getMore: getMoreStub
     });
     const session = sinon.createStubInstance(ClientSession);
-    const operation = new GetMoreOperation(ns, cursorId, server, options);
+    const opts = { ...options, session };
+    const operation = new GetMoreOperation(ns, cursorId, server, opts);
 
     it('executes a getmore on the provided server', function (done) {
       const callback = () => {
         const call = getMoreStub.getCall(0);
         expect(getMoreStub.calledOnce).to.be.true;
         expect(call.args[0]).to.equal(ns);
         expect(call.args[1]).to.equal(cursorId);
-        expect(call.args[2]).to.deep.equal({ ...options, session });
+        expect(call.args[2]).to.deep.equal(opts);
         done();
       };
       operation.execute(server, session, callback);