Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

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

Merged
Merged
Show file tree
Hide file tree
Changes from 10 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions packages/i18n/src/locales/en_US.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2771,6 +2771,9 @@ const translations: Catalog = {
description:
'Return stats captured from a named stream processor.',
},
modify: {
description: 'Modify a stream processor definition.',
},
},
},
},
Expand Down
38 changes: 38 additions & 0 deletions packages/shell-api/src/stream-processor.ts
Original file line number Diff line number Diff line change
@@ -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';
Expand Down Expand Up @@ -57,6 +58,43 @@ 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 = {}
addaleax marked this conversation as resolved.
Show resolved Hide resolved
) {
if (Array.isArray(pipelineOrOptions)) {
options['pipeline'] = pipelineOrOptions;
addaleax marked this conversation as resolved.
Show resolved Hide resolved
} else if (typeof pipelineOrOptions === 'object') {
if (Object.keys(options).length !== 0) {
addaleax marked this conversation as resolved.
Show resolved Hide resolved
throw new MongoshInvalidInputError(
'If the first argument to modify is an object, the second argument should not be specified.',
CommonErrors.InvalidArgument
);
}
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({
Expand Down
109 changes: 109 additions & 0 deletions packages/shell-api/src/streams.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import type Mongo from './mongo';
import Database from './database';
import { Streams } from './streams';
import { InterruptFlag, MongoshInterruptedError } from './interruptor';
import type { MongoshInvalidInputError } from '@mongosh/errors';

describe('Streams', function () {
let mongo: Mongo;
Expand Down Expand Up @@ -162,4 +163,112 @@ 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.'
);

// Two object arguments to modify.
const caught3 = await processor
.modify(
{ resumeFromCheckpoint: false },
{ dlq: { connectionName: 'foo' } }
)
.catch((e: MongoshInvalidInputError) => e);
expect(caught3.message).to.contain(
'[COMMON-10001] If the first argument to modify is an object, the second argument should not be specified.'
);
});

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;
});
});
});
Loading