feat(shell-api): add stream processor modify MONGOSH-1864

packages/shell-api/src/stream-processor.ts

@@ -1,4 +1,5 @@
 import type { Document } from '@mongosh/service-provider-core';
+import { CommonErrors, MongoshInvalidInputError } from '@mongosh/errors';
 
 import type Mongo from './mongo';
 import { asPrintable } from './enums';
@@ -57,6 +58,37 @@ export default class StreamProcessor extends ShellApiWithMongoClass {
     });
   }
 
+  /**
+   * modify can be used like so:
+   *  Change the pipeline:
+   *    sp.name.modify(pipeline)
+   *  Change the pipeline with additional options:
+   *    sp.name.modify(pipeline, {resumeFromCheckpoint: false})
+   *  For modify requests that don't change the pipeline:
+   *    sp.name.modify({resumeFromCheckpoint: false})
+   */
+  @returnsPromise
+  async modify(
+    pipelineOrOptions: Document[] | Document,
+    options: Document = {}
+  ) {
+    if (Array.isArray(pipelineOrOptions)) {
+      options['pipeline'] = pipelineOrOptions;
+    } else if (typeof pipelineOrOptions === 'object') {
+      options = { ...options, ...pipelineOrOptions };
+    } else {
+      throw new MongoshInvalidInputError(
+        'The first argument to modify must be an array or object.',
+        CommonErrors.InvalidArgument
+      );
+    }
+
+    return this._streams._runStreamCommand({
+      modifyStreamProcessor: this.name,
+      ...options,
+    });
+  }
+
   @returnsPromise
   async sample(options: Document = {}) {
     const r = await this._streams._runStreamCommand({

packages/shell-api/src/streams.spec.ts

@@ -5,6 +5,7 @@ import type Mongo from './mongo';
 import Database from './database';
 import { Streams } from './streams';
 import { InterruptFlag, MongoshInterruptedError } from './interruptor';
+import { MongoshInvalidInputError } from '@mongosh/errors';
 
 describe('Streams', function () {
   let mongo: Mongo;
@@ -162,4 +163,101 @@ describe('Streams', function () {
       ).to.be.true;
     });
   });
+
+  describe('modify', function () {
+    it('throws with invalid parameters', async function () {
+      // Create the stream processor.
+      const runCmdStub = sinon
+        .stub(mongo._serviceProvider, 'runCommand')
+        .resolves({ ok: 1 });
+      const name = 'p1';
+      const pipeline = [{ $match: { foo: 'bar' } }];
+      const processor = await streams.createStreamProcessor(name, pipeline);
+      expect(processor).to.eql(streams.getProcessor(name));
+      const cmd = { createStreamProcessor: name, pipeline };
+      expect(runCmdStub.calledOnceWithExactly('admin', cmd, {})).to.be.true;
+
+      // No arguments to modify.
+      const caught = await processor
+        .modify()
+        .catch((e: MongoshInvalidInputError) => e);
+      expect(caught.message).to.contain(
+        '[COMMON-10001] The first argument to modify must be an array or object.'
+      );
+
+      // A single numeric argument to modify.
+      const caught2 = await processor
+        .modify(1)
+        .catch((e: MongoshInvalidInputError) => e);
+      expect(caught2.message).to.contain(
+        '[COMMON-10001] The first argument to modify must be an array or object.'
+      );
+    });
+
+    it('works with pipeline and options arguments', async function () {
+      const runCmdStub = sinon
+        .stub(mongo._serviceProvider, 'runCommand')
+        .resolves({ ok: 1 });
+
+      // Create the stream processor.
+      const name = 'p1';
+      const pipeline = [{ $match: { foo: 'bar' } }];
+      const processor = await streams.createStreamProcessor(name, pipeline);
+      expect(processor).to.eql(streams.getProcessor(name));
+      const cmd = { createStreamProcessor: name, pipeline };
+      expect(runCmdStub.calledOnceWithExactly('admin', cmd, {})).to.be.true;
+
+      // Start the stream processor.
+      await processor.start();
+      expect(
+        runCmdStub.calledWithExactly(
+          'admin',
+          { startStreamProcessor: name },
+          {}
+        )
+      ).to.be.true;
+
+      // Stop the stream processor.
+      await processor.stop();
+      expect(
+        runCmdStub.calledWithExactly('admin', { stopStreamProcessor: name }, {})
+      ).to.be.true;
+
+      // Modify the stream processor.
+      const pipeline2 = [{ $match: { foo: 'baz' } }];
+      processor.modify(pipeline2);
+      expect(
+        runCmdStub.calledWithExactly(
+          'admin',
+          { modifyStreamProcessor: name, pipeline: pipeline2 },
+          {}
+        )
+      ).to.be.true;
+
+      // Modify the stream processor with extra options.
+      const pipeline3 = [{ $match: { foo: 'bat' } }];
+      processor.modify(pipeline3, { resumeFromCheckpoint: false });
+      expect(
+        runCmdStub.calledWithExactly(
+          'admin',
+          {
+            modifyStreamProcessor: name,
+            pipeline: pipeline3,
+            resumeFromCheckpoint: false,
+          },
+          {}
+        )
+      ).to.be.true;
+
+      // Modify the stream processor without changing pipeline.
+      processor.modify({ resumeFromCheckpoint: false });
+      expect(
+        runCmdStub.calledWithExactly(
+          'admin',
+          { modifyStreamProcessor: name, resumeFromCheckpoint: false },
+          {}
+        )
+      ).to.be.true;
+    });
+  });
 });