[MongoDB] Change Stream: Watch

watch는 MongoDB에서 CDC를 구현할 때 사용할 수 있는 기능 중 하나다.

특정 컬렉션, 또는 데이터베이스 단위에 Stream을 열어서 변경사항이 발생할때마다 그 변경사항을 전달받고 대응할 수 있게 하는 것이다.

단일 노드에는 사용이 불가능하고, replicaset이나 클러스터 모드에서만 사용 가능한 기능이다.

예제코드는 Node.js다. 다른 언어라도 사용 원리에 있어서 크게 다른 점은 없을 것이다.

먼저 당연히 클라이언트 객체를 생성해야 한다.

그러고 watch를 걸고싶은 컬렉션 객체까지 가져온 다음에, watch 함수를 호출해서 stream을 열면 그만이다.

const changeStream = db.weather.watch();
changeStream.on('change', next => {
  console.log(next);
});

이렇게 콜백 형태로 해서 변경사항을 받아서 처리하면 된다.
매번 데이터가 변경될때마다 저 change 콜백으로 데이터가 날라올 것이다.

그리고 실행해둔 상태에서

데이터를 삽입해보면

바로바로 날라올 것이다.

DB 전체에 대해서 트리거를 거는 것도 방법은 같다.
그냥 db에서 watch 메서드를 부르면 되는 것이다.

const { MongoClient } = require("mongodb");

const url =
  "mongodb+srv://sssang...ngodb.net";
const client = new MongoClient(url);

async function main() {
  await client.connect();
  console.log("Connected successfully to server");

  const db = client.db("test");

  const collection = db.collection("weather");

  const changeStream = collection.watch();

  changeStream.on("change", (next) => {
    console.log(next);
  });

  // wait
  await new Promise((resolve) => setTimeout(resolve, 60000));
}

main()
  .then(() => {
    console.log("Done");
    process.exit(0);
  })
  .catch((err) => {
    console.error(err);
    process.exit(1);
  });

그리고 아무 컬렉션에나 데이터를 쑤셔보면

마찬가지로 데이터가 날라올 것이다.

참조
https://www.mongodb.com/docs/manual/changeStreams/#std-label-changeStreams
https://www.mongodb.com/docs/manual/reference/method/db.collection.watch/