changestream

changeStream.js

const { MongoClient } = require("mongodb");
// Replace the uri string with your MongoDB deployment's connection string.
const uri = "mongodb+srv://U:P@cD?writeConcern=majority";

const client = new MongoClient(uri);
let changeStream;
async function run() {
    try {
        await client.connect();
        const database = client.db("sample_mflix");
        const collection = database.collection("movies");
        // open a Change Stream on the "movies" collection
        changeStream = collection.watch();
        // set up a listener when change events are emitted
        changeStream.on("change", next => {
            // process any change event
            console.log("received a change to the collection: \t", next);
        });
        // use a timeout to ensure the listener is registered before the insertOne
        // operation is called.
        await new Promise(resolve => {
            setTimeout(async () => {
                await collection.insertOne({
                    test: "sample movie document",
                });
            }, 1000);
        });
    } finally {
        await client.close();
    }
}
run().catch(console.dir);

changeStreamWithResumablity.js

const { MongoClient } = require("mongodb");
const uri = "mongodb+srv://U:P@D?writeConcern=majority";

const test = async () => {
    console.log("Initializing");
    const client = new MongoClient(uri);

    await client.connect();
    const database = client.db("sample_mflix");
    const collection = database.collection("movies");
 
    let newChangeStream = collection.watch({ resumeAfter: {"_data":"8260218058000000012B022C0100296E5A10042EDE84BCC5A54CEAAC03A00226C5A42346645F69640064602180569CC60730BBE4FCE40004"}});
    newChangeStream.on('change', next => {
        console.log(next);
    });

}

test();