MongoDB 更改流 returns 插入时清空 fullDocument

Question

Mongo 4.4 和相应的 Golang 驱动程序被使用。数据库的副本集正在运行本地 localhost:27017、localhost:27020。我也尝试过使用 Atlas 的沙箱集群，它给了我相同的结果。

根据 Mongo 的 documentation 在处理插入新文档时 fullDocument 事件数据字段应该包含新插入的文档，但由于某种原因并非如此为了我。 ns 应该是数据库和集合名称的字段以及存储受影响的文档 _id 的 documentKey 字段也是空的。 operationType 字段包含正确的操作类型。在另一项测试中，似乎更新操作根本没有出现在更改流中。

它过去可以正常工作，但现在不正常了。为什么会发生，我做错了什么？

代码

// ds is the connection to discord, required for doing stuff inside handlers
func iterateChangeStream(stream *mongo.ChangeStream, ds *discordgo.Session, ctx context.Context, cancel context.CancelFunc) {
    defer stream.Close(ctx)
    defer cancel() // for graceful crashing

    for stream.Next(ctx) {
        var event bson.M
        err := stream.Decode(&event)
        if err != nil {
            log.Print(errors.Errorf("Failed to decode event: %w\n", err))
            return
        }

        rv := reflect.ValueOf(event["operationType"]) // getting operation type
        opType, ok := rv.Interface().(string)
        if !ok {
            log.Print("String expected in operationType\n")
            return
        }
        
        // event["fullDocument"] will be empty even when handling insertion
        // models.Player is a struct representing a document of the collection
        // I'm watching over
        doc, ok := event["fullDocument"].(models.Player)
        if !ok {
            log.Print("Failed to convert document into Player type")
            return
        }
        handlerCtx := context.WithValue(ctx, "doc", doc)
        // handlerToEvent maps operationType to respective handler
        go handlerToEvent[opType](ds, handlerCtx, cancel)
    }
}

func WatchEvents(ds *discordgo.Session, ctx context.Context, cancel context.CancelFunc) {

    pipeline := mongo.Pipeline{
        bson.D{{
            "$match",
            bson.D{{
                "$or", bson.A{
                    bson.D{{"operationType", "insert"}}, // !!!
                    bson.D{{"operationType", "delete"}},
                    bson.D{{"operationType", "invalidate"}},
                },
            }},
        }},
    }
    // mongo instance is initialized on program startup and stored in a global variable
    opts := options.ChangeStream().SetFullDocument(options.UpdateLookup)
    stream, err := db.Instance.Collection.Watch(ctx, pipeline, opts)
    if err != nil {
        log.Panic(err)
    }
    defer stream.Close(ctx)

    iterateChangeStream(stream, ds, ctx, cancel)
}

我的问题可能与 this 有关，只不过它始终发生在插入时，而不是有时发生在更新时。如果您知道如何启用上面 link 中提到的更改流优化功能标志，请告诉我。

随时询问更多说明。

Answer 1

从我从 this link 看到的示例事件中，我们可以看到 fullDocument 仅存在于 operationType: 'insert'。

 { 
     _id: { _data: '825DE67A42000000072B022C0100296E5A10046BBC1C6A9CBB4B6E9CA9447925E693EF46645F696400645DE67A42113EA7DE6472E7680004' },
    operationType: 'insert',
    clusterTime: Timestamp { _bsontype: 'Timestamp', low_: 7, high_: 1575385666 },
    fullDocument: { 
        _id: 5de67a42113ea7de6472e768,
        name: 'Sydney Harbour Home',
        bedrooms: 4,
        bathrooms: 2.5,
        address: { market: 'Sydney', country: 'Australia' } },
        ns: { db: 'sample_airbnb', coll: 'listingsAndReviews' },
        documentKey: { _id: 5de67a42113ea7de6472e768 } 
 }
 { 
    _id: { _data: '825DE67A42000000082B022C0100296E5A10046BBC1C6A9CBB4B6E9CA9447925E693EF46645F696400645DE67A42113EA7DE6472E7680004' },
    operationType: 'delete',
    clusterTime: Timestamp { _bsontype: 'Timestamp', low_: 8, high_: 1575385666 },
    ns: { db: 'sample_airbnb', coll: 'listingsAndReviews' },
    documentKey: { _id: 5de67a42113ea7de6472e768 } 
 }

所以我推荐你

将您的 $match 限制为 insert
或将 if 语句添加到 operationType。

      if opType == "insert" {
        doc, ok := event["fullDocument"].(models.Player)
        if !ok {
            log.Print("Failed to convert document into Player type")
            return
        }
        handlerCtx := context.WithValue(ctx, "doc", doc)
        // handlerToEvent maps operationType to respective handler
        go handlerToEvent[opType](ds, handlerCtx, cancel)
        return
      }

或确保您正在使用来自 event["documentKey"]["_id"] 的文档 ID 获取文档并调用 playersCollection.findOne({_id: event["documentKey"]["_id"]})

Answer 2

问题已回答here。

TLDR

您需要创建以下结构以将事件解组为：

type CSEvent struct {
    OperationType string        `bson:"operationType"`
    FullDocument  models.Player `bson:"fullDocument"`
}

var event CSEvent
err := stream.Decode(&event)

event 将包含插入文档的副本。

MongoDB 更改流 returns 插入时清空 fullDocument

MongoDB change stream returns empty fullDocument on insert

go

mongodb

changestream

代码

TLDR