I am currently working on a gateway service that basically converts mqtt topic publications into a different representation that is then sent on gRPC streams. To hold the different topics and their values (json serialized objects) in-memory, we use hashicorps memdb.

One reason for choosing memdb is the ability to register a watcher for table changes. To put it simply, the published value is written into the corresponding table, which then triggers a reaction via the registered watcher.

The code looks something like this:

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
}

What I don't understand is the fact that we have to create a new watcher every time the watcher was triggered. If the watcher is not reinitialized, ws.WatchCtx() is not blocking anymore. My guess is that we have misunderstood the concept and are therefore using the tools incorrectly. So my question is how to use memdb watchsets.