データの変更を監視

Overview

このガイドでは、変更ストリームを使用してデータに対するリアルタイムの変更を監視する方法を学習できます。 変更ストリームは、アプリケーションがコレクション、データベース、または配置のデータ変更をサブスクライブできる MongoDB Server の機能です。

Scalaドライバーを使用する場合、watch() メソッドを呼び出して ChangeStreamObservable のインスタンスを返すことができます。その後、ChangeStreamObservableインスタンスをサブスクライブして、アップデート、挿入、削除などの新しいデータ変更を確認できます。

val database: MongoDatabase = client.getDatabase("sample_restaurants")
val collection: MongoCollection[Document] = database.getCollection("restaurants")

case class LatchedObserver() extends Observer[ChangeStreamDocument[Document]] {
  val latch = new CountDownLatch(1)
  override def onSubscribe(subscription: Subscription): Unit = subscription.request(Long.MaxValue) // Request data
  override def onNext(changeDocument: ChangeStreamDocument[Document]): Unit = println(changeDocument)
  override def onError(throwable: Throwable): Unit = {
      println(s"Error: '$throwable")
      latch.countDown()
  }
  override def onComplete(): Unit = latch.countDown()
  def await(): Unit = latch.await()
}

変更ストリームを開く

変更ストリームを開くには、 watch()メソッドを呼び出します。 watch()メソッドを呼び出すインスタンスによって、変更ストリームが監視するイベントの範囲が決まります。 次のクラスのインスタンスでwatch()メソッドを呼び出すことができます。

• MongoClient: 配置内のすべてのデータベースにわたるすべてのコレクションに対する変更を監視します。ただし、システム コレクションまたは admin、 、 データベースのコレクションは対象外です。localconfig

• MongoDatabase: 1 つのデータベース内のすべてのコレクションに対する変更を監視

• MongoCollection: 1 つのコレクションの変更を監視

次の例では、 watch() メソッドを呼び出して、restaurantsコレクションの変更ストリームを開きます。 このコードでは、変更が発生したときに受信して出力する LatchedObserverインスタンスが作成されます。

val observer = LatchedObserver()
collection.watch().subscribe(observer)
observer.await()

変更の監視を開始するには、上記のコードを実行します。 次に、別のシェルで次のコードを実行して、nameフィールドの値が "Blarney Castle" であるドキュメントを更新します。

val filter = equal("name", "Blarney Castle")
val update = set("cuisine", "American")
collection.updateOne(filter, update)
          .subscribe((res: UpdateResult) => println(res),
                    (e: Throwable) => println(s"There was an error: $e"))

上記のコードを実行してコレクション を更新すると、変更ストリームアプリケーションは変更が発生に応じて出力します。 出力される変更イベントは、次の出力のようになります。

ChangeStreamDocument{ operationType=update, resumeToken={"_data": "..."},
namespace=sample_restaurants.restaurants, destinationNamespace=null,
fullDocument=null, fullDocumentBeforeChange=null, documentKey={"_id": {...}},
clusterTime=Timestamp{...}, updateDescription=UpdateDescription{removedFields=[],
updatedFields={"cuisine": "Irish"}, truncatedArrays=[], disambiguatedPaths=null},
txnNumber=null, lsid=null, splitEvent=null, wallTime=BsonDateTime{...}}

変更ストリーム出力の変更

変更ストリーム出力を変更するには、パイプラインステージのリストをパラメーターとして watch() メソッドに渡します。 リストには、次のステージを含めることができます。

• $addFields または$set : ドキュメントに新しいフィールドを追加します

• $match: ドキュメントをフィルタリングします

• $project:ドキュメントフィールドのサブセットをプロジェクションします

• $replaceWith または$replaceRoot : 入力ドキュメントを指定されたドキュメントで置き換え

• $redact: ドキュメントのコンテンツを制限します

• $unset: ドキュメントからフィールドを削除します

Scalaドライバーは Aggregatesクラスを提供します。このクラスには、前述のパイプラインステージを構築するためのヘルパーメソッドが含まれています。

次の例では、 Aggregates.filter() メソッドを使用して $match ステージを構築するパイプラインを作成します。 次に、コードはこのパイプラインをwatch() メソッドに渡し、アップデート操作が発生した場合にのみイベントを出力するように watch() に指示します。

val observer = LatchedObserver()
collection.watch(Seq(Aggregates.filter(Filters.in("operationType", "update"))))
observer.await()

watch() の動作を変更する

ChangeStreamObservableクラスによって提供されるメソッドを連鎖させることで、watch() メソッドの動作を変更できます。 次の表では、これらの方法の一部について説明しています。

watch() オプションの完全なリストについては、 APIドキュメントのChangeStreamObservableを参照してください。

val observer = LatchedObserver()
collection.watch()
          .fullDocument(FullDocument.UPDATE_LOOKUP)
          .subscribe(observer)
observer.await()

ChangeStreamDocument{ operationType=update, resumeToken={"_data": "..."},
namespace=sample_restaurants.restaurants, destinationNamespace=null,
fullDocument=Iterable((_id,BsonObjectId{...}), (address,{"building": "202-24",
"coord": [-73.9250442, 40.5595462], "street": "Rockaway Point Boulevard",
"zipcode": "11697"}), (borough,BsonString{value='Queens'}),
(cuisine,BsonString{value='Irish'}), (grades,BsonArray{values=[...]}),
(name,BsonString{value='Blarney Castle'}), (restaurant_id,BsonString{...}),
(blank,BsonString{value='Irish'})), fullDocumentBeforeChange=null,
documentKey={"_id": {"$oid": "..."}}, clusterTime=Timestamp{...}, updateDescription=
UpdateDescription{removedFields=[], updatedFields={"cuisine": "Irish"},
truncatedArrays=[], disambiguatedPaths=null}, txnNumber=null, lsid=null,
splitEvent=null, wallTime=BsonDateTime{...}}

詳細情報

Change Streams変更ストリームの詳細については、MongoDB Server マニュアルの 「 ストリーム」 を参照してください。