Skip to content

Commit 673e28d

Browse files
mbroadstmbroadst
committed
remove isClosed() method, depend on closed accessor
1 parent a8ec062 commit 673e28d

File tree

9 files changed

+46
-45
lines changed

9 files changed

+46
-45
lines changed

‎package-lock.json

Lines changed: 2 additions & 1 deletion
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎src/change_stream.ts

Lines changed: 17 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@ import { executeOperation, ExecutionResult } from './operations/execute_operatio
3131

3232
const kResumeQueue = Symbol('resumeQueue');
3333
const kCursorStream = Symbol('cursorStream');
34+
const kClosed = Symbol('closed');
3435

3536
const CHANGE_STREAM_OPTIONS = ['resumeAfter', 'startAfter', 'startAtOperationTime', 'fullDocument'];
3637
const CURSOR_OPTIONS = ['batchSize', 'maxAwaitTimeMS', 'collation', 'readPreference'].concat(
@@ -180,10 +181,10 @@ export class ChangeStream extends EventEmitter {
180181
namespace: MongoDBNamespace;
181182
type: symbol;
182183
cursor?: ChangeStreamCursor;
183-
closed: boolean;
184184
streamOptions?: CursorStreamOptions;
185185
[kResumeQueue]: Denque;
186186
[kCursorStream]?: Readable;
187+
[kClosed]: boolean;
187188

188189
/** @event */
189190
static readonly CLOSE = 'close' as const;
@@ -241,7 +242,7 @@ export class ChangeStream extends EventEmitter {
241242
// Create contained Change Stream cursor
242243
this.cursor = createChangeStreamCursor(this, options);
243244

244-
this.closed = false;
245+
this[kClosed] = false;
245246

246247
// Listen for any `change` listeners being added to ChangeStream
247248
this.on('newListener', eventName => {
@@ -296,23 +297,20 @@ export class ChangeStream extends EventEmitter {
296297
}
297298

298299
/** Is the cursor closed */
299-
isClosed(): boolean {
300-
return this.closed || (this.cursor?.isClosed() ?? false);
300+
get closed(): boolean {
301+
return this[kClosed] || (this.cursor?.closed ?? false);
301302
}
302303

303304
/** Close the Change Stream */
304305
close(callback?: Callback): Promise<void> | void {
305-
return maybePromise(callback, cb => {
306-
if (this.closed) return cb();
307-
308-
// flag the change stream as explicitly closed
309-
this.closed = true;
306+
this[kClosed] = true;
310307

311-
if (!this.cursor) return cb();
308+
return maybePromise(callback, cb => {
309+
if (!this.cursor) {
310+
return cb();
311+
}
312312

313-
// Tidy up the existing cursor
314313
const cursor = this.cursor;
315-
316314
return cursor.close(err => {
317315
endStream(this);
318316
this.cursor = undefined;
@@ -584,7 +582,7 @@ function processNewChange(
584582
change: ChangeStreamDocument,
585583
callback?: Callback
586584
) {
587-
if (changeStream.closed) {
585+
if (changeStream[kClosed]) {
588586
if (callback) callback(CHANGESTREAM_CLOSED_ERROR);
589587
return;
590588
}
@@ -614,8 +612,8 @@ function processError(changeStream: ChangeStream, error: AnyError, callback?: Ca
614612
const cursor = changeStream.cursor;
615613

616614
// If the change stream has been closed explicitly, do not process error.
617-
if (changeStream.closed) {
618-
if (callback) callback(new MongoError('ChangeStream is closed'));
615+
if (changeStream[kClosed]) {
616+
if (callback) callback(CHANGESTREAM_CLOSED_ERROR);
619617
return;
620618
}
621619

@@ -674,8 +672,8 @@ function processError(changeStream: ChangeStream, error: AnyError, callback?: Ca
674672
* @param changeStream - the parent ChangeStream
675673
*/
676674
function getCursor(changeStream: ChangeStream, callback: Callback<ChangeStreamCursor>) {
677-
if (changeStream.isClosed()) {
678-
callback(new MongoError('ChangeStream is closed.'));
675+
if (changeStream[kClosed]) {
676+
callback(CHANGESTREAM_CLOSED_ERROR);
679677
return;
680678
}
681679

@@ -698,8 +696,8 @@ function getCursor(changeStream: ChangeStream, callback: Callback<ChangeStreamCu
698696
function processResumeQueue(changeStream: ChangeStream, err?: Error) {
699697
while (changeStream[kResumeQueue].length) {
700698
const request = changeStream[kResumeQueue].pop();
701-
if (changeStream.isClosed() && !err) {
702-
request(new MongoError('Change Stream is not open.'));
699+
if (changeStream[kClosed] && !err) {
700+
request(CHANGESTREAM_CLOSED_ERROR);
703701
return;
704702
}
705703

‎src/cursor/abstract_cursor.ts

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -161,11 +161,6 @@ export abstract class AbstractCursor extends EventEmitter {
161161
return this[kKilled];
162162
}
163163

164-
// NOTE: should we remove these? They are currently needed by a number of tests
165-
isClosed(): boolean {
166-
return this.closed;
167-
}
168-
169164
/** Returns current buffered documents length */
170165
bufferedCount(): number {
171166
return this[kDocuments].length;

‎src/index.ts

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
11
import { Instrumentation } from './apm';
2+
import { AbstractCursor } from './cursor/abstract_cursor';
23
import { AggregationCursor } from './cursor/aggregation_cursor';
4+
import { FindCursor } from './cursor/find_cursor';
5+
import { ListIndexesCursor } from './operations/indexes';
6+
import { ListCollectionsCursor } from './operations/list_collections';
37
import { PromiseProvider } from './promise_provider';
48
import { Admin } from './admin';
59
import { MongoClient } from './mongo_client';
@@ -69,7 +73,11 @@ export {
6973
Collection,
7074
ReadPreference,
7175
Logger,
76+
AbstractCursor,
7277
AggregationCursor,
78+
FindCursor,
79+
ListIndexesCursor,
80+
ListCollectionsCursor,
7381
GridFSBucket
7482
};
7583

‎src/operations/list_collections.ts

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -102,6 +102,7 @@ export class ListCollectionsOperation extends CommandOperation<ListCollectionsOp
102102
}
103103
}
104104

105+
/** @public */
105106
export class ListCollectionsCursor extends AbstractCursor {
106107
parent: Db;
107108
filter: Document;

‎src/sdam/monitor.ts

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,7 @@ export class Monitor extends EventEmitter {
6464
[kServer]: Server;
6565
[kConnection]?: Connection;
6666
[kCancellationToken]: EventEmitter;
67+
/** @internal */
6768
[kMonitorId]?: InterruptableAsyncInterval;
6869
[kRTTPinger]?: RTTPinger;
6970

‎test/functional/change_stream.test.js

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -475,14 +475,14 @@ describe('Change Streams', function () {
475475
const changeStream = database.collection('changeStreamCloseTest').watch(pipeline);
476476
this.defer(() => changeStream.close());
477477

478-
assert.equal(changeStream.isClosed(), false);
479-
assert.equal(changeStream.cursor.isClosed(), false);
478+
assert.equal(changeStream.closed, false);
479+
assert.equal(changeStream.cursor.closed, false);
480480

481481
changeStream.close(err => {
482482
expect(err).to.not.exist;
483483

484484
// Check the cursor is closed
485-
assert.equal(changeStream.isClosed(), true);
485+
assert.equal(changeStream.closed, true);
486486
assert.ok(!changeStream.cursor);
487487
done();
488488
});
@@ -774,7 +774,7 @@ describe('Change Streams', function () {
774774
changeStream.hasNext(function (err, hasNext) {
775775
expect(err).to.not.exist;
776776
assert.equal(hasNext, false);
777-
assert.equal(changeStream.isClosed(), true);
777+
assert.equal(changeStream.closed, true);
778778
done();
779779
});
780780
}
@@ -841,7 +841,7 @@ describe('Change Streams', function () {
841841
.then(() => changeStream.hasNext())
842842
.then(function (hasNext) {
843843
assert.equal(hasNext, false);
844-
assert.equal(changeStream.isClosed(), true);
844+
assert.equal(changeStream.closed, true);
845845
done();
846846
});
847847
});
@@ -1866,7 +1866,7 @@ describe('Change Streams', function () {
18661866
afterEach(function () {
18671867
return Promise.resolve()
18681868
.then(() => {
1869-
if (changeStream && !changeStream.isClosed()) {
1869+
if (changeStream && !changeStream.closed) {
18701870
return changeStream.close();
18711871
}
18721872
})

‎test/functional/cursor.test.js

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -927,7 +927,7 @@ describe('Cursor', function () {
927927
cursor.next((err, items) => {
928928
expect(err).to.not.exist;
929929
test.ok(items == null);
930-
test.ok(cursor.isClosed());
930+
test.ok(cursor.closed);
931931
done();
932932
});
933933
});
@@ -993,7 +993,7 @@ describe('Cursor', function () {
993993
cursor.next((err, items) => {
994994
expect(err).to.not.exist;
995995
test.ok(items == null);
996-
test.ok(cursor.isClosed());
996+
test.ok(cursor.closed);
997997
done();
998998
});
999999
});
@@ -1050,7 +1050,7 @@ describe('Cursor', function () {
10501050
cursor.next((err, items) => {
10511051
expect(err).to.not.exist;
10521052
test.ok(items == null);
1053-
test.ok(cursor.isClosed());
1053+
test.ok(cursor.closed);
10541054
done();
10551055
});
10561056
});
@@ -1208,7 +1208,7 @@ describe('Cursor', function () {
12081208
const cursor = collection.find();
12091209
cursor.close(err => {
12101210
expect(err).to.not.exist;
1211-
test.equal(true, cursor.isClosed());
1211+
test.equal(true, cursor.closed);
12121212
done();
12131213
});
12141214
});
@@ -1409,7 +1409,7 @@ describe('Cursor', function () {
14091409

14101410
cursor.close(err => {
14111411
expect(err).to.not.exist;
1412-
test.equal(true, cursor.isClosed());
1412+
test.equal(true, cursor.closed);
14131413
done();
14141414
});
14151415
});
@@ -1623,7 +1623,7 @@ describe('Cursor', function () {
16231623
test.equal(1, closed);
16241624
test.equal(1, paused);
16251625
test.equal(1, resumed);
1626-
test.strictEqual(cursor.isClosed(), true);
1626+
test.strictEqual(cursor.closed, true);
16271627
done();
16281628
}
16291629
});
@@ -1679,7 +1679,7 @@ describe('Cursor', function () {
16791679
if (doneCalled === 1) {
16801680
expect(err).to.not.exist;
16811681
test.strictEqual(0, i);
1682-
test.strictEqual(true, cursor.isClosed());
1682+
test.strictEqual(true, cursor.closed);
16831683
done();
16841684
}
16851685
};
@@ -1722,7 +1722,7 @@ describe('Cursor', function () {
17221722
const cursor = collection.find();
17231723
const stream = cursor.stream();
17241724

1725-
test.strictEqual(false, cursor.isClosed());
1725+
test.strictEqual(false, cursor.closed);
17261726

17271727
stream.on('data', function () {
17281728
if (++i === 5) {
@@ -1740,7 +1740,7 @@ describe('Cursor', function () {
17401740
test.strictEqual(undefined, err);
17411741
test.strictEqual(5, i);
17421742
test.strictEqual(2, finished);
1743-
test.strictEqual(true, cursor.isClosed());
1743+
test.strictEqual(true, cursor.closed);
17441744
done();
17451745
}
17461746
}
@@ -1797,7 +1797,7 @@ describe('Cursor', function () {
17971797
if (finished === 2) {
17981798
setTimeout(function () {
17991799
test.equal(5, i);
1800-
test.equal(true, cursor.isClosed());
1800+
test.equal(true, cursor.closed);
18011801
client.close();
18021802

18031803
configuration.manager.start().then(function () {

‎test/functional/operation_example.test.js

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -5454,9 +5454,6 @@ describe('Operation Examples', function () {
54545454

54555455
/**
54565456
* A simple example showing the use of the cursor close function.
5457-
*
5458-
* @example-class Cursor
5459-
* @example-method isClosed
54605457
*/
54615458
it('shouldStreamDocumentsUsingTheIsCloseFunction', {
54625459
// Add a tag that our runner can trigger on
@@ -5505,7 +5502,7 @@ describe('Operation Examples', function () {
55055502
// Close the cursor, this is the same as reseting the query
55065503
cursor.close(function (err) {
55075504
expect(err).to.not.exist;
5508-
test.equal(true, cursor.isClosed());
5505+
test.equal(true, cursor.closed);
55095506

55105507
client.close(done);
55115508
});

0 commit comments

Comments
 (0)