change_streams.js

/* eslint no-unused-vars: 0 */
'use strict';

const setupDatabase = require('../functional/shared').setupDatabase;
const expect = require('chai').expect;

describe('examples(change-stream):', function() {
  let client;
  let db;

  before(async function() {
    await setupDatabase(this.configuration);
  });

  beforeEach(async function() {
    client = await this.configuration.newClient().connect();
    db = client.db(this.configuration.db);

    await db.createCollection('inventory');
    await db.collection('inventory').deleteMany({});
  });

  afterEach(async function() {
    await client.close();
    client = undefined;
    db = undefined;
  });

  class Looper {
    constructor(lambda, interval) {
      this._run = false;
      this._lambda = lambda;
      this._interval = interval || 50;
    }

    async _go() {
      this._run = true;
      while (this._run) {
        await new Promise(r => setTimeout(r, this._interval));
        await this._lambda();
      }
    }

    run() {
      this._p = this._go().catch(() => {});
    }

    stop() {
      this._run = false;
      return this._p;
    }
  }

  it('Open A Change Stream', {
    metadata: { requires: { topology: ['replicaset'], mongodb: '>=3.6.0' } },
    test: async function() {
      const looper = new Looper(() => db.collection('inventory').insertOne({ a: 1 }));
      looper.run();

      // Start Changestream Example 1
      const collection = db.collection('inventory');
      const changeStream = collection.watch();
      changeStream.on('change', next => {
        // process next document
      });
      // End Changestream Example 1

      // Start Changestream Example 1 Alternative
      const changeStreamIterator = collection.watch();
      const next = await changeStreamIterator.next();
      // End Changestream Example 1 Alternative

      await changeStream.close();
      await changeStreamIterator.close();
      await looper.stop();

      expect(next)
        .to.have.property('operationType')
        .that.equals('insert');
    }
  });

  it('Lookup Full Document for Update Operations', {
    metadata: { requires: { topology: ['replicaset'], mongodb: '>=3.6.0' } },
    test: async function() {
      await db.collection('inventory').insertOne({ a: 1, b: 2 });
      const looper = new Looper(() =>
        db.collection('inventory').updateOne({ a: 1 }, { $set: { a: 2 } })
      );
      looper.run();

      // Start Changestream Example 2
      const collection = db.collection('inventory');
      const changeStream = collection.watch({ fullDocument: 'updateLookup' });
      changeStream.on('change', next => {
        // process next document
      });
      // End Changestream Example 2

      // Start Changestream Example 2 Alternative
      const changeStreamIterator = collection.watch({ fullDocument: 'updateLookup' });
      const next = await changeStreamIterator.next();
      // End Changestream Example 2 Alternative

      await changeStream.close();
      await changeStreamIterator.close();
      await looper.stop();

      expect(next)
        .to.have.property('operationType')
        .that.equals('update');
      expect(next)
        .to.have.property('fullDocument')
        .that.has.all.keys(['_id', 'a', 'b']);
    }
  });

  it('Resume a Change Stream', {
    metadata: { requires: { topology: ['replicaset'], mongodb: '>=3.6.0' } },
    test: async function() {
      const looper = new Looper(async () => {
        await db.collection('inventory').insertOne({ a: 1 });
        await db.collection('inventory').insertOne({ b: 2 });
      });
      looper.run();

      // Start Changestream Example 3
      const collection = db.collection('inventory');
      const changeStream = collection.watch();

      let newChangeStream;
      changeStream.on('change', next => {
        const resumeToken = changeStream.resumeToken;
        changeStream.close();

        newChangeStream = collection.watch({ resumeAfter: resumeToken });
        newChangeStream.on('change', next => {
          // process next document
        });
      });
      // End Changestream Example 3

      // Start Changestream Example 3 Alternative
      const changeStreamIterator = collection.watch();
      const change1 = await changeStreamIterator.next();

      const resumeToken = changeStreamIterator.resumeToken;
      changeStreamIterator.close();

      const newChangeStreamIterator = collection.watch({ resumeAfter: resumeToken });
      const change2 = await newChangeStreamIterator.next();
      // End Changestream Example 3 Alternative

      await newChangeStreamIterator.close();
      await newChangeStream.close();
      await looper.stop();

      expect(change1).to.have.nested.property('fullDocument.a', 1);
      expect(change2).to.have.nested.property('fullDocument.b', 2);
    }
  });

  it('Modify Change Stream Output', {
    metadata: { requires: { topology: ['replicaset'], mongodb: '>=3.6.0' } },
    test: async function() {
      const looper = new Looper(async () => {
        await db.collection('inventory').insertOne({ username: 'alice' });
      });
      looper.run();

      // Start Changestream Example 4
      const pipeline = [
        { $match: { 'fullDocument.username': 'alice' } },
        { $addFields: { newField: 'this is an added field!' } }
      ];

      const collection = db.collection('inventory');
      const changeStream = collection.watch(pipeline);
      changeStream.on('change', next => {
        // process next document
      });
      // End Changestream Example 4

      // Start Changestream Example 4 Alternative
      const changeStreamIterator = collection.watch(pipeline);
      const next = await changeStreamIterator.next();
      // End Changestream Example 4 Alternative

      await changeStream.close();
      await changeStreamIterator.close();
      await looper.stop();

      expect(next).to.have.nested.property('fullDocument.username', 'alice');
      expect(next).to.have.property('newField', 'this is an added field!');
    }
  });
});