Monitorar alterações de dados

Visão geral

Neste guia, você pode aprender como usar um fluxo de alterações para monitorar alterações em tempo real em seus dados. Um change stream é uma funcionalidade do MongoDB Server que permite que seu aplicação se inscreva em alterações de dados em uma collection, banco de dados de dados ou sistema.

Ao utilizar o driver Scala, você pode chamar o método watch() para retornar uma instância do ChangeStreamObservable. Em seguida, você pode assinar a instância do ChangeStreamObservable para ver novas alterações de dados, como atualizações, inserções e exclusões.

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()
}

Abrir um fluxo de alterações

Para abrir um fluxo de alteração, chame o método watch() . A instância na qual você chama o método watch() determina o escopo de eventos que o change stream monitora. Você pode chamar o método watch() em instâncias das seguintes classes:

• MongoClient: monitora alterações em todas as collections em todos os bancos de dados em um sistema, excluindo as collections do sistema ou as collections nos admin local config bancos de dados, e

• MongoDatabase: Monitora alterações em todas as coleções em um banco de dados

• MongoCollection: monitora alterações em uma coleção

O exemplo a seguir chama o método watch() para abrir um fluxo de alteração na coleção restaurants. O código cria uma instância do LatchedObserver para receber e produzir alterações conforme ocorrem:

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

Para começar a observar as alterações, execute o código anterior. Em seguida, em uma shell separada, execute o seguinte código para atualizar um documento que tenha um valor de campo name de "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"))

Quando você executa o código anterior para atualizar a coleção, o aplicação de fluxo de alterações imprime a alteração conforme ela ocorre. O evento de alteração impresso se assemelha à seguinte saída:

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{...}}

Modificar a saída change stream

Para modificar a saída do change stream, passe uma lista de estágios do pipeline como parâmetro para o método watch(). Você pode incluir os seguintes estágios na lista:

• $addFields ou $set: adiciona novos campos aos documentos

• $match: filtra os documentos

• $project: projeta um subconjunto dos campos do documento

• $replaceWith ou $replaceRoot: substitui o documento de entrada pelo documento especificado

• $redact: restringe o conteúdo dos documentos

• $unset: remove campos de documentos

O driver Scala fornece a classe Aggregates , que inclui métodos de assistente para criar os estágios anteriores do pipeline.

O exemplo a seguir cria um pipeline que usa o método Aggregates.filter() para construir o estágio $match. Em seguida, o código passa esse pipeline para o método watch() e instrui watch() a gerar eventos somente quando ocorrerem operações de atualização:

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

Modificar comportamento do watch()

Você pode modificar o comportamento do método watch() encadeando métodos fornecidos pela classe ChangeStreamObservable . A tabela a seguir descreve alguns desses métodos:

Para obter uma lista completa das watch() opções, consulte ChangeStreamObservable na documentação da API.

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{...}}

Informações adicionais

Para saber mais sobre fluxos de alterações, consulte Change Streams de alterações no manual do MongoDB Server .