130 lines
3.2 KiB
Go
130 lines
3.2 KiB
Go
package link
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
)
|
|
|
|
// KVUpdate is one change of a key in a bucket the node-engine watches (novox/hq ADR 0539 §9). Caught marks the end of
|
|
// the keys that were there when the watch began.
|
|
type KVUpdate struct {
|
|
Key string
|
|
Value []byte
|
|
Revision uint64
|
|
Deleted bool
|
|
Caught bool
|
|
}
|
|
|
|
// KVBus is a link that can watch a bucket's keys: the node agent follows its wants and its compositions with it, and
|
|
// never polls.
|
|
type KVBus interface {
|
|
// KVWatch yields the keys of the bucket matching the pattern now, a Caught update, then every change, until ctx ends
|
|
// or the link does (the channel is then closed). A bucket the node's account may not read is an error.
|
|
KVWatch(ctx context.Context, bucket, pattern string) (<-chan KVUpdate, error)
|
|
}
|
|
|
|
// KVWatch is the watch on the link open now.
|
|
func (b OverNATS) KVWatch(ctx context.Context, bucket, pattern string) (<-chan KVUpdate, error) {
|
|
js := b.JS
|
|
if b.Conn != nil {
|
|
bound, err := b.Conn.JetStream()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
js = bound
|
|
}
|
|
kv, err := js.KeyValue(bucket)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("the bucket %s: %w", bucket, err)
|
|
}
|
|
w, err := kv.Watch(pattern, nats.Context(ctx))
|
|
if err != nil {
|
|
return nil, fmt.Errorf("watching %s in %s: %w", pattern, bucket, err)
|
|
}
|
|
out := make(chan KVUpdate, 64)
|
|
go func() {
|
|
defer close(out)
|
|
defer func() { _ = w.Stop() }()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case e, ok := <-w.Updates():
|
|
if !ok {
|
|
return
|
|
}
|
|
var u KVUpdate
|
|
if e == nil {
|
|
u.Caught = true
|
|
} else {
|
|
u = KVUpdate{Key: e.Key(), Value: e.Value(), Revision: e.Revision(),
|
|
Deleted: e.Operation() == nats.KeyValueDelete || e.Operation() == nats.KeyValuePurge}
|
|
}
|
|
select {
|
|
case out <- u:
|
|
case <-ctx.Done():
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
return out, nil
|
|
}
|
|
|
|
func (l *natsLink) KVWatch(ctx context.Context, bucket, pattern string) (<-chan KVUpdate, error) {
|
|
return OverNATS{Conn: l.conn, JS: l.js}.KVWatch(ctx, bucket, pattern)
|
|
}
|
|
|
|
var (
|
|
_ KVBus = OverNATS{}
|
|
_ KVBus = (*natsLink)(nil)
|
|
)
|
|
|
|
// ErrNoLink is what a watch asks for with no link open.
|
|
var ErrNoLink = errors.New("no link to the bus is open")
|
|
|
|
// KVWatch watches a bucket on the link open now. It ends when ctx does or the link closes.
|
|
func (q *Session) KVWatch(ctx context.Context, bucket, pattern string) (<-chan KVUpdate, error) {
|
|
q.init()
|
|
q.mu.Lock()
|
|
bus := q.bus
|
|
q.mu.Unlock()
|
|
w, ok := bus.(KVBus)
|
|
if bus == nil || !ok {
|
|
return nil, ErrNoLink
|
|
}
|
|
return w.KVWatch(ctx, bucket, pattern)
|
|
}
|
|
|
|
// WhenLinked waits for a link to be open and returns a context that ends when that link does (or ctx does): what the
|
|
// node agent runs under, so its watches end with the link and begin again with the next.
|
|
func (q *Session) WhenLinked(ctx context.Context) (context.Context, context.CancelFunc, error) {
|
|
q.init()
|
|
for {
|
|
q.mu.Lock()
|
|
if q.bus != nil {
|
|
gone := q.gone
|
|
q.mu.Unlock()
|
|
linked, cancel := context.WithCancel(ctx)
|
|
go func() {
|
|
select {
|
|
case <-gone:
|
|
cancel()
|
|
case <-linked.Done():
|
|
}
|
|
}()
|
|
return linked, cancel, nil
|
|
}
|
|
changed := q.changed
|
|
q.mu.Unlock()
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil, nil, ctx.Err()
|
|
case <-changed:
|
|
}
|
|
}
|
|
}
|