如何使用memdb.WatchSet.WatchCtx(ctx)？

Question

我目前正在开发一个网关服务，该服务基本上将 mqtt 主题出版物转换为不同的表示形式，然后在 gRPC 流上发送。为了将不同的主题及其值（json 序列化对象）保存在内存中，我们使用 hashcorps memdb。

选择 memdb 的原因之一是能够注册表更改的观察程序。简单来说，就是将发布的值写入相应的表中，然后通过注册的观察者触发反应。

代码看起来像这样：

func main() {
    ..
    go handle(ctx)
}

func handle(ctx context.Context) {
    if ws, al, err := Watch(); err != nil {
        // log
    } else {

        for {
            // blocking call according to documentation
            if err = ws.WatchCtx(ctx); err != nil {
                fmt.Println("received cancel, exit loop")
                break
            } else {
                fmt.Println("received update")

                // reinit watcher to prevent being retriggered by the same event
                if ws, al, err = Watch(); err != nil {
                    // log
                    break
                } else {
                    fmt.Println(al)
                }
            }
        }
    }
}

func Watch() (ws memdb.WatchSet, al AccessList, err error) {
    txn := db.Txn(false)

    if wc, v, e := txn.FirstWatch(accessTable, idIndex, id); e != nil {
        err = e
    } else if v == nil {
        err = memdb.ErrNotFound
    } else {
        ws = memdb.NewWatchSet()
        ws.Add(wc)
        al = v.(AccessList)
    }

    return
}

我不明白的是，每次触发观察者时我们都必须创建一个新的观察者。如果观察者没有重新初始化，ws.WatchCtx() 就不再阻塞。 我的猜测是我们误解了这个概念，因此错误地使用了这些工具。 所以我的问题是如何使用 memdb 手表集。