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.

Sign up for GitHub

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

Jump to bottom

fix(NODE-3521): update session support checks #3151

Merged
merged 12 commits into from Mar 3, 2022
Merged

fix(NODE-3521): update session support checks #3151

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
a0dfc5b
fix(NODE-3521): remove session support checks
durran Feb 24, 2022
0f8cd6e
fix(NODE-3521): remove extra server selection
durran Feb 25, 2022
ef6894d
Revert "fix(NODE-3521): remove extra server selection"
durran Feb 28, 2022
b0ca0b6
Revert "fix(NODE-3521): remove session support checks"
durran Feb 28, 2022
d3e2109
fix(NODE-3521): properly check session check
durran Feb 28, 2022
ddc4d24
test(NODE-3521): add next tests for session
durran Mar 1, 2022
dc62173
chore(NODE-3521): find/agg cursor session always there
durran Mar 1, 2022
5839dae
test(NODE-3521): fix cursor server selection tests
durran Mar 2, 2022
68365bf
Update test/unit/cursor/aggregation_cursor.test.js
durran Mar 2, 2022
dd594f6
Update test/unit/cursor/aggregation_cursor.test.js
durran Mar 2, 2022
05ad2d2
Update test/unit/cursor/find_cursor.test.js
durran Mar 2, 2022
8ec9ca0
fix(NODE-3521): lint errors
durran Mar 3, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
11 changes: 9 additions & 2 deletions src/cursor/abstract_cursor.ts
View file
Open in desktop
Expand Up @@ -653,8 +653,15 @@ function next<T>(cursor: AbstractCursor, blocking: boolean, callback: Callback<T

if (cursorId == null) {
// All cursors must operate within a session, one must be made implicitly if not explicitly provided
if (cursor[kSession] == null && cursor[kTopology].hasSessionSupport()) {
cursor[kSession] = cursor[kTopology].startSession({ owner: cursor, explicit: false });
if (cursor[kSession] == null) {
if (cursor[kTopology].shouldCheckForSessionSupport()) {
nbbeeken marked this conversation as resolved.
Show resolved Hide resolved
return cursor[kTopology].selectServer(ReadPreference.primaryPreferred, err => {
durran marked this conversation as resolved.
Show resolved Hide resolved
if (err) return callback(err);
return next(cursor, blocking, callback);
});
} else if (cursor[kTopology].hasSessionSupport()) {
cursor[kSession] = cursor[kTopology].startSession({ owner: cursor, explicit: false });
}
}

cursor._initialize(cursor[kSession], (err, state) => {
nbbeeken marked this conversation as resolved.
Show resolved Hide resolved
Expand Down
2 changes: 1 addition & 1 deletion src/cursor/aggregation_cursor.ts
View file
Open in desktop
Expand Up @@ -61,7 +61,7 @@ export class AggregationCursor<TSchema = Document> extends AbstractCursor<TSchem
}

/** @internal */
_initialize(session: ClientSession | undefined, callback: Callback<ExecutionResult>): void {
_initialize(session: ClientSession, callback: Callback<ExecutionResult>): void {
const aggregateOperation = new AggregateOperation(this.namespace, this[kPipeline], {
...this[kOptions],
...this.cursorOptions,
Expand Down
2 changes: 1 addition & 1 deletion src/cursor/find_cursor.ts
View file
Open in desktop
Expand Up @@ -68,7 +68,7 @@ export class FindCursor<TSchema = Document> extends AbstractCursor<TSchema> {
}

/** @internal */
_initialize(session: ClientSession | undefined, callback: Callback<ExecutionResult>): void {
_initialize(session: ClientSession, callback: Callback<ExecutionResult>): void {
const findOperation = new FindOperation(undefined, this.namespace, this[kFilter], {
...this[kBuiltOptions], // NOTE: order matters here, we may need to refine this
...this.cursorOptions,
Expand Down
138 changes: 138 additions & 0 deletions test/unit/cursor/aggregation_cursor.test.js
View file
Open in desktop
@@ -0,0 +1,138 @@
'use strict';

const { expect } = require('chai');
const mock = require('../../tools/mongodb-mock/index');
const { Topology } = require('../../../src/sdam/topology');
const { Long } = require('bson');
const { MongoDBNamespace, isHello } = require('../../../src/utils');
const { AggregationCursor } = require('../../../src/cursor/aggregation_cursor');

const test = {};
describe('Aggregation Cursor', function () {
describe('#next', function () {
afterEach(function () {
mock.cleanup();
});
beforeEach(async function () {
test.server = await mock.createServer();
});

context('when there is a data bearing server', function () {
beforeEach(function () {
test.server.setMessageHandler(request => {
const doc = request.document;
if (isHello(doc)) {
request.reply(mock.HELLO);
} else if (doc.aggregate) {
request.reply({
cursor: {
id: Long.fromNumber(1),
ns: 'test.test',
firstBatch: [{ _id: 1, name: 'test' }]
},
ok: 1
});
}
});
});

it('sets the session on the cursor', function (done) {
const topology = new Topology(test.server.hostAddress());
const cursor = new AggregationCursor(
topology,
MongoDBNamespace.fromString('test.test'),
[],
{}
);
topology.connect(function () {
cursor.next(function () {
expect(cursor.session).to.exist;
topology.close(done);
});
});
});
});

context('when there is no data bearing server', function () {
beforeEach(function () {
test.server.setMessageHandler(request => {
const doc = request.document;
if (isHello(doc)) {
request.reply({ errmsg: 'network error' });
} else if (doc.aggregate) {
request.reply({
cursor: {
id: Long.fromNumber(1),
ns: 'test.test',
firstBatch: [{ _id: 1, name: 'test' }]
},
ok: 1
});
}
});
});

it('does not set the session on the cursor', function (done) {
const topology = new Topology(test.server.hostAddress(), {
serverSelectionTimeoutMS: 1000
});
const cursor = new AggregationCursor(
topology,
MongoDBNamespace.fromString('test.test'),
[],
{}
);
topology.connect(function () {
cursor.next(function () {
expect(cursor.session).to.not.exist;
topology.close(done);
});
});
});
});

context('when a data bearing server becomes available', function () {
beforeEach(function () {
// Set the count of times hello has been called.
let helloCalls = 0;
test.server.setMessageHandler(request => {
const doc = request.document;
if (isHello(doc)) {
// After the first hello call errors indicating no data bearing server is
// available, any subsequent hello call should succeed after server selection.
// This gives us a data bearing server available for the next call.
request.reply(helloCalls > 0 ? mock.HELLO : { errmsg: 'network error' });
helloCalls++;
} else if (doc.aggregate) {
request.reply({
cursor: {
id: Long.fromNumber(1),
ns: 'test.test',
firstBatch: [{ _id: 1, name: 'test' }]
},
ok: 1
});
}
});
});

it('sets the session on the cursor', function (done) {
const topology = new Topology(test.server.hostAddress(), {
serverSelectionTimeoutMS: 1000
});
const cursor = new AggregationCursor(
topology,
MongoDBNamespace.fromString('test.test'),
[],
{}
);
topology.connect(function () {
cursor.next(function () {
expect(cursor.session).to.exist;
topology.close(done);
});
});
});
});
});
});
112 changes: 112 additions & 0 deletions test/unit/cursor/find_cursor.test.js
View file
Open in desktop
Expand Up @@ -10,6 +10,118 @@ const { FindCursor } = require('../../../src/cursor/find_cursor');

const test = {};
describe('Find Cursor', function () {
describe('#next', function () {
afterEach(function () {
mock.cleanup();
});
beforeEach(async function () {
test.server = await mock.createServer();
});

context('when there is a data bearing server', function () {
beforeEach(function () {
test.server.setMessageHandler(request => {
const doc = request.document;
if (isHello(doc)) {
request.reply(mock.HELLO);
} else if (doc.find) {
request.reply({
cursor: {
id: Long.fromNumber(1),
ns: 'test.test',
firstBatch: [{ _id: 1, name: 'test' }]
},
ok: 1
});
}
});
});

it('sets the session on the cursor', function (done) {
const topology = new Topology(test.server.hostAddress());
const cursor = new FindCursor(topology, MongoDBNamespace.fromString('test.test'), {}, {});
topology.connect(function () {
cursor.next(function () {
expect(cursor.session).to.exist;
topology.close(done);
});
});
});
});

context('when there is no data bearing server', function () {
beforeEach(function () {
test.server.setMessageHandler(request => {
const doc = request.document;
if (isHello(doc)) {
request.reply({ errmsg: 'network error' });
} else if (doc.find) {
request.reply({
cursor: {
id: Long.fromNumber(1),
ns: 'test.test',
firstBatch: [{ _id: 1, name: 'test' }]
},
ok: 1
});
}
});
});

it('does not set the session on the cursor', function (done) {
const topology = new Topology(test.server.hostAddress(), {
serverSelectionTimeoutMS: 1000
});
const cursor = new FindCursor(topology, MongoDBNamespace.fromString('test.test'), {}, {});
topology.connect(function () {
cursor.next(function () {
expect(cursor.session).to.not.exist;
topology.close(done);
});
});
});
});

context('when a data bearing server becomes available', function () {
beforeEach(function () {
// Set the count of times hello has been called.
let helloCalls = 0;
test.server.setMessageHandler(request => {
const doc = request.document;
if (isHello(doc)) {
// After the first hello call errors indicating no data bearing server is
// available, any subsequent hello call should succeed after server selection.
// This gives us a data bearing server available for the next call.
request.reply(helloCalls > 0 ? mock.HELLO : { errmsg: 'network error' });
helloCalls++;
} else if (doc.find) {
request.reply({
cursor: {
id: Long.fromNumber(1),
ns: 'test.test',
firstBatch: [{ _id: 1, name: 'test' }]
},
ok: 1
});
}
});
});

it('sets the session on the cursor', function (done) {
const topology = new Topology(test.server.hostAddress(), {
serverSelectionTimeoutMS: 1000
});
const cursor = new FindCursor(topology, MongoDBNamespace.fromString('test.test'), {}, {});
topology.connect(function () {
cursor.next(function () {
expect(cursor.session).to.exist;
topology.close(done);
});
});
});
});
});

describe('Response', function () {
afterEach(() => mock.cleanup());
beforeEach(() => {
Expand Down