changestream.js

const client = new MongoClient(MONGODB_CONNECTION_STRING);
await client.connect();
const collection = client.db("myapp").collection("outbox");

/* 
watch() function accepts an aggregation pipeline, which can be used to perform
additional aggregation stages. Check out MongoDB aggregation framework for more info:
https://docs.mongodb.com/manual/core/aggregation-pipeline/
*/
const changeStream = collection.watch([
  {
    $match: { operationType: "insert" },
  },
]);
changeStream.on("change", (changeEvent) => {
  /* 
  dispatchToBroker function will perform any transformation
  to the change event before sending it to the message broker 
  */
  dispatchToBroker(changeEvent)
});