MongoDb C# 驱动程序过滤器 FullDocument 字段上的 WatchAsync

Question

我正在尝试过滤集合更改，仅选择满足特定过滤器的更新、删除和插入的文档。 旧式过滤器（在 oplog 上工作）是

  // filter generated by settings
  FilterDefinition<BsonDocument> filter = GetFilterFromOverriddenMethod();
  // interested only in u, d and filtered i documents
  filter = Builders<BsonDocument>.Filter.Or(
    Builders<BsonDocument>.Filter.In("op", new[] { "u", "d" }),
    Builders<BsonDocument>.Filter.And(
      Builders<BsonDocument>.Filter.Eq("op", "i"),
      filter
    )
  );
  // filter applied on oplog collection
  filter = Builders<BsonDocument>.Filter.And(
    Builders<BsonDocument>.Filter.Eq("ns", "dbName.collectionName",
    Builders<BsonDocument>.Filter.Gt("ts", lastId),
    filter
  );

我正在尝试使用 collection.WatchAsync 重现（或多或少）相同的行为，并且无法修改 GetFilterFromOverriddenMethod。 我尝试使用管道

  FilterDefinition<BsonDocument> filter = GetFilterFromOverriddenMethod();
  var deleteUpdateFilter = Builders<ChangeStreamDocument<BsonDocument>>.Filter.In(change => 
        change.OperationType, 
        new[] { ChangeStreamOperationType.Update, ChangeStreamOperationType.Delete }
        );
  var insertReplaceFilter = Builders<ChangeStreamDocument<BsonDocument>>.Filter.In(change => 
        change.OperationType, 
        new[] { ChangeStreamOperationType.Insert, ChangeStreamOperationType.Replace }
        );
  var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<BsonDocument>>()
    .Match( deleteUpdateFilter |
      (insertReplaceFilter &
        Builders<ChangeStreamDocument<BsonDocument>>.Filter.**??**(change => change.FullDocument, filter)
      )
    );

但我无法集成过滤器（

FilterDefinition<BsonDocument>

不适用于

FilterDefinition<ChangeStreamDocument<BsonDocument>>

）。 我还尝试打印管道以破解其细节并从序列化管道生成过滤器，但我无法打印

FilterDefinition<ChangeStreamDocument<BsonDocument>>

，

  var documentSerializer = BsonSerializer.SerializerRegistry.GetSerializer<ChangeStreamDocument<BsonDocument>>();

  var renderedFilter = deleteUpdateFilter.Render(documentSerializer, BsonSerializer.SerializerRegistry);
                Console.WriteLine(renderedFilter.ToString());

抛出调用 GetSerializer 的异常

System.MissingMethodException: No parameterless constructor defined for type 'MongoDB.Driver.ChangeStreamDocumentSerializer`1[MongoDB.Bson.BsonDocument]'.

有人有什么建议吗？

Answer 1

找了一些资料，有需要的可以分享：

1. 无法使用流保留为 oplog 生成的过滤器，因为 oplog 过滤器必须明确引用 oplog 项的“o”字段，而 ChangeStream 接口中不存在
2. filter 可以使用不同的前缀，指的是“fullDocument”（过滤器变成像

   Builders<BsonDocument>.Filter.Eq("fullDocument.foo", true)

   ）
3. 可以在

   FilterDefinition<T>

   中组合

   FilterDefinition<U>

   渲染它并创建 BsonDocumentFilterDefinition，例如

    var filter = Builders<BsonDocument>.Filter("fullDocument.foo", true);
    var documentSerializer = BsonSerializer.SerializerRegistry.GetSerializer<BsonDocument>();
    var renderedFilter = filter.Render(documentSerializer, BsonSerializer.SerializerRegistry);
    var subFilter = new BsonDocumentFilterDefinition<ChangeStreamDocument<BsonDocument>>(renderedFilter);

    var pipelineFilter = 
      Builders<ChangeStreamDocument<BsonDocument>>.Filter.In(
        change => change.OperationType, new[] { ChangeStreamOperationType.Update, ChangeStreamOperationType.Delete }
      ) |
      (
        Builders<ChangeStreamDocument<BsonDocument>>.Filter.In(
          change => change.OperationType, new[] { ChangeStreamOperationType.Insert, ChangeStreamOperationType.Replace }
        ) &
        subFilter
      );