|
1 | 1 | 'use strict';
|
2 | 2 | const assert = require('assert');
|
3 | 3 | const { Transform, PassThrough } = require('stream');
|
4 |
| -const { MongoNetworkError, MongoDriverError } = require('../../src/error'); |
| 4 | +const { MongoNetworkError } = require('../../src/error'); |
5 | 5 | const { delay, setupDatabase, withClient, withCursor } = require('./shared');
|
6 | 6 | const co = require('co');
|
7 | 7 | const mock = require('../tools/mock');
|
8 |
| -const { EventCollector } = require('../tools/utils'); |
| 8 | +const { EventCollector, getSymbolFrom } = require('../tools/utils'); |
9 | 9 | const chai = require('chai');
|
10 | 10 | const expect = chai.expect;
|
11 | 11 | const sinon = require('sinon');
|
@@ -99,12 +99,7 @@ function triggerResumableError(changeStream, delay, onClose) {
|
99 | 99 | * @param {Function} callback
|
100 | 100 | */
|
101 | 101 | function waitForStarted(changeStream, callback) {
|
102 |
| - const timeout = setTimeout(() => { |
103 |
| - expect.fail('Change stream never started'); |
104 |
| - }, 2000); |
105 |
| - |
106 | 102 | changeStream.cursor.once('init', () => {
|
107 |
| - clearTimeout(timeout); |
108 | 103 | callback();
|
109 | 104 | });
|
110 | 105 | }
|
@@ -176,26 +171,25 @@ const pipeline = [
|
176 | 171 | ];
|
177 | 172 |
|
178 | 173 | describe('Change Streams', function () {
|
179 |
| - before(function () { |
180 |
| - return setupDatabase(this.configuration, ['integration_tests']); |
| 174 | + before(async function () { |
| 175 | + return await setupDatabase(this.configuration, ['integration_tests']); |
181 | 176 | });
|
182 | 177 |
|
183 |
| - beforeEach(function () { |
| 178 | + beforeEach(async function () { |
184 | 179 | const configuration = this.configuration;
|
185 | 180 | const client = configuration.newClient();
|
186 | 181 |
|
187 |
| - return client |
188 |
| - .connect() |
189 |
| - .then(() => { |
190 |
| - const db = client.db('integration_tests'); |
191 |
| - return db.createCollection('test'); |
192 |
| - }) |
193 |
| - .then( |
194 |
| - () => client.close(), |
195 |
| - () => client.close() |
196 |
| - ); |
| 182 | + await client.connect(); |
| 183 | + const db = client.db('integration_tests'); |
| 184 | + try { |
| 185 | + await db.createCollection('test'); |
| 186 | + } catch { |
| 187 | + // ns already exists, don't care |
| 188 | + } finally { |
| 189 | + await client.close(); |
| 190 | + } |
197 | 191 | });
|
198 |
| - afterEach(() => mock.cleanup()); |
| 192 | + afterEach(async () => await mock.cleanup()); |
199 | 193 |
|
200 | 194 | it('should close the listeners after the cursor is closed', {
|
201 | 195 | metadata: { requires: { topology: 'replicaset', mongodb: '>=3.6' } },
|
@@ -580,39 +574,39 @@ describe('Change Streams', function () {
|
580 | 574 | }
|
581 | 575 | });
|
582 | 576 |
|
583 |
| - it( |
584 |
| - 'should error if resume token projected out of change stream document using imperative callback form', |
585 |
| - { |
586 |
| - metadata: { requires: { topology: 'replicaset', mongodb: '>=3.6' } }, |
| 577 | + it('should error if resume token projected out of change stream document using iterator', { |
| 578 | + metadata: { requires: { topology: 'replicaset', mongodb: '>=3.6' } }, |
| 579 | + test(done) { |
| 580 | + const configuration = this.configuration; |
| 581 | + const client = configuration.newClient(); |
587 | 582 |
|
588 |
| - test: function (done) { |
589 |
| - const configuration = this.configuration; |
590 |
| - const client = configuration.newClient(); |
| 583 | + client.connect((err, client) => { |
| 584 | + expect(err).to.not.exist; |
591 | 585 |
|
592 |
| - client.connect((err, client) => { |
593 |
| - expect(err).to.not.exist; |
594 |
| - this.defer(() => client.close()); |
| 586 | + const database = client.db('integration_tests'); |
| 587 | + const collection = database.collection('resumetokenProjectedOutCallback'); |
| 588 | + const changeStream = collection.watch([{ $project: { _id: false } }]); |
595 | 589 |
|
596 |
| - const database = client.db('integration_tests'); |
597 |
| - const changeStream = database |
598 |
| - .collection('resumetokenProjectedOutCallback') |
599 |
| - .watch([{ $project: { _id: false } }]); |
600 |
| - this.defer(() => changeStream.close()); |
| 590 | + changeStream.hasNext(() => {}); // trigger initialize |
601 | 591 |
|
602 |
| - // Trigger the first database event |
603 |
| - waitForStarted(changeStream, () => { |
604 |
| - this.defer(database.collection('resumetokenProjectedOutCallback').insert({ b: 2 })); |
605 |
| - }); |
| 592 | + changeStream.cursor.on('init', () => { |
| 593 | + collection.insertOne({ b: 2 }, (err, res) => { |
| 594 | + expect(err).to.be.undefined; |
| 595 | + expect(res).to.exist; |
606 | 596 |
|
607 |
| - // Fetch the change notification |
608 |
| - changeStream.next(err => { |
609 |
| - expect(err).to.exist; |
610 |
| - done(); |
| 597 | + changeStream.next(err => { |
| 598 | + expect(err).to.exist; |
| 599 | + changeStream.close(() => { |
| 600 | + client.close(() => { |
| 601 | + done(); |
| 602 | + }); |
| 603 | + }); |
| 604 | + }); |
611 | 605 | });
|
612 | 606 | });
|
613 |
| - } |
| 607 | + }); |
614 | 608 | }
|
615 |
| - ); |
| 609 | + }); |
616 | 610 |
|
617 | 611 | it('should error if resume token projected out of change stream document using event listeners', {
|
618 | 612 | metadata: { requires: { topology: 'replicaset', mongodb: '>=3.6' } },
|
@@ -1792,109 +1786,86 @@ describe('Change Streams', function () {
|
1792 | 1786 | }
|
1793 | 1787 | });
|
1794 | 1788 |
|
1795 |
| - // FIXME: NODE-1797 |
1796 | 1789 | describe('should error when used as iterator and emitter concurrently', function () {
|
1797 |
| - let client, coll, changeStream, repeatInsert, val; |
1798 |
| - val = 0; |
| 1790 | + let client, coll, changeStream, kMode, initPromise; |
1799 | 1791 |
|
1800 | 1792 | beforeEach(async function () {
|
1801 | 1793 | client = this.configuration.newClient();
|
1802 |
| - await client.connect().catch(() => expect.fail('Failed to connect to client')); |
| 1794 | + await client.connect(); |
1803 | 1795 |
|
1804 | 1796 | coll = client.db(this.configuration.db).collection('tester');
|
1805 | 1797 | changeStream = coll.watch();
|
1806 |
| - |
1807 |
| - repeatInsert = setInterval(async function () { |
1808 |
| - await coll.insertOne({ c: val }).catch('Failed to insert document'); |
1809 |
| - val++; |
1810 |
| - }, 75); |
| 1798 | + kMode = getSymbolFrom(changeStream, 'mode'); |
| 1799 | + initPromise = new Promise(resolve => waitForStarted(changeStream, resolve)); |
1811 | 1800 | });
|
1812 | 1801 |
|
1813 | 1802 | afterEach(async function () {
|
1814 |
| - if (repeatInsert) { |
1815 |
| - clearInterval(repeatInsert); |
1816 |
| - } |
| 1803 | + let err; |
1817 | 1804 | if (changeStream) {
|
1818 |
| - await changeStream.close(); |
| 1805 | + try { |
| 1806 | + if (changeStream[kMode] === 'emitter') { |
| 1807 | + // shutting down the client will end the session, if this happens before |
| 1808 | + // the stream initialization aggregate operation is processed, it will throw |
| 1809 | + // a session ended error, which can't be caught if we end the stream, so |
| 1810 | + // we need to wait for the stream to initialize before closing all the things |
| 1811 | + await initPromise; |
| 1812 | + } |
| 1813 | + await changeStream.close(); |
| 1814 | + } catch (error) { |
| 1815 | + // don't throw before closing the client |
| 1816 | + err = error; |
| 1817 | + } |
1819 | 1818 | }
|
1820 | 1819 |
|
1821 |
| - await mock.cleanup(); |
1822 | 1820 | if (client) {
|
1823 | 1821 | await client.close();
|
1824 | 1822 | }
|
1825 |
| - }); |
1826 | 1823 |
|
1827 |
| - it( |
1828 |
| - 'should throw MongoDriverError when set as an emitter with "on" and used as an iterator with "hasNext"', |
1829 |
| - { |
1830 |
| - metadata: { requires: { topology: 'replicaset', mongodb: '>=3.6' } }, |
1831 |
| - test: async function () { |
1832 |
| - await new Promise(resolve => changeStream.on('change', resolve)); |
1833 |
| - try { |
1834 |
| - await changeStream.hasNext().catch(err => { |
1835 |
| - expect.fail(err.message); |
1836 |
| - }); |
1837 |
| - } catch (error) { |
1838 |
| - return expect(error).to.be.instanceof(MongoDriverError); |
1839 |
| - } |
1840 |
| - return expect.fail('Should not reach here'); |
1841 |
| - } |
| 1824 | + if (err) { |
| 1825 | + throw err; |
1842 | 1826 | }
|
1843 |
| - ); |
1844 |
| - |
1845 |
| - it( |
1846 |
| - 'should throw MongoDriverError when set as an iterator with "hasNext" and used as an emitter with "on"', |
1847 |
| - { |
1848 |
| - metadata: { requires: { topology: 'replicaset', mongodb: '>=3.6' } }, |
1849 |
| - test: async function () { |
1850 |
| - await changeStream |
1851 |
| - .hasNext() |
1852 |
| - .catch(() => expect.fail('Failed to set changeStream to iterator')); |
1853 |
| - try { |
1854 |
| - await new Promise(resolve => changeStream.on('change', resolve)); |
1855 |
| - } catch (error) { |
1856 |
| - return expect(error).to.be.instanceof(MongoDriverError); |
1857 |
| - } |
1858 |
| - return expect.fail('Should not reach here'); |
1859 |
| - } |
1860 |
| - } |
1861 |
| - ); |
1862 |
| - |
1863 |
| - it( |
1864 |
| - 'should throw MongoDriverError when set as an emitter with "once" and used as an iterator with "next"', |
1865 |
| - { |
1866 |
| - metadata: { requires: { topology: 'replicaset', mongodb: '>=3.6' } }, |
1867 |
| - test: async function () { |
1868 |
| - await new Promise(resolve => changeStream.once('change', resolve)); |
1869 |
| - try { |
1870 |
| - await changeStream.next().catch(err => { |
1871 |
| - expect.fail(err.message); |
1872 |
| - }); |
1873 |
| - } catch (error) { |
1874 |
| - return expect(error).to.be.instanceof(MongoDriverError); |
1875 |
| - } |
1876 |
| - return expect.fail('Should not reach here'); |
1877 |
| - } |
| 1827 | + }); |
| 1828 | + |
| 1829 | + it(`should throw when mixing event listeners with iterator methods`, { |
| 1830 | + metadata: { requires: { topology: 'replicaset', mongodb: '>=3.6' } }, |
| 1831 | + async test() { |
| 1832 | + expect(changeStream).to.have.property(kMode, false); |
| 1833 | + // ChangeStream detects emitter usage via 'newListener' event |
| 1834 | + // so this covers all emitter methods |
| 1835 | + changeStream.on('change', () => {}); |
| 1836 | + expect(changeStream).to.have.property(kMode, 'emitter'); |
| 1837 | + |
| 1838 | + const errRegex = /ChangeStream cannot be used as an iterator/; |
| 1839 | + |
| 1840 | + // These all throw synchronously so it should be safe to not await the results |
| 1841 | + expect(() => { |
| 1842 | + changeStream.next(); |
| 1843 | + }).to.throw(errRegex); |
| 1844 | + expect(() => { |
| 1845 | + changeStream.hasNext(); |
| 1846 | + }).to.throw(errRegex); |
| 1847 | + expect(() => { |
| 1848 | + changeStream.tryNext(); |
| 1849 | + }).to.throw(errRegex); |
1878 | 1850 | }
|
1879 |
| - ); |
1880 |
| - |
1881 |
| - it( |
1882 |
| - 'should throw MongoDriverError when set as an iterator with "tryNext" and used as an emitter with "on"', |
1883 |
| - { |
1884 |
| - metadata: { requires: { topology: 'replicaset', mongodb: '>=3.6' } }, |
1885 |
| - test: async function () { |
1886 |
| - await changeStream |
1887 |
| - .tryNext() |
1888 |
| - .catch(() => expect.fail('Failed to set changeStream to iterator')); |
1889 |
| - try { |
1890 |
| - await new Promise(resolve => changeStream.on('change', resolve)); |
1891 |
| - } catch (error) { |
1892 |
| - return expect(error).to.be.instanceof(MongoDriverError); |
1893 |
| - } |
1894 |
| - return expect.fail('Should not reach here'); |
1895 |
| - } |
| 1851 | + }); |
| 1852 | + |
| 1853 | + it(`should throw when mixing iterator methods with event listeners`, { |
| 1854 | + metadata: { requires: { topology: 'replicaset', mongodb: '>=3.6' } }, |
| 1855 | + async test() { |
| 1856 | + expect(changeStream).to.have.property(kMode, false); |
| 1857 | + const res = await changeStream.tryNext(); |
| 1858 | + expect(res).to.not.exist; |
| 1859 | + expect(changeStream).to.have.property(kMode, 'iterator'); |
| 1860 | + |
| 1861 | + // This does throw synchronously |
| 1862 | + // the newListener event is called sync |
| 1863 | + // which calls streamEvents, which calls setIsEmitter, which will throw |
| 1864 | + expect(() => { |
| 1865 | + changeStream.on('change', () => {}); |
| 1866 | + }).to.throw(/ChangeStream cannot be used as an EventEmitter/); |
1896 | 1867 | }
|
1897 |
| - ); |
| 1868 | + }); |
1898 | 1869 | });
|
1899 | 1870 |
|
1900 | 1871 | describe('should properly handle a changeStream event being processed mid-close', function () {
|
|
0 commit comments