72 lines
2.6 KiB
Go
72 lines
2.6 KiB
Go
package process
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/membership"
|
|
)
|
|
|
|
// FollowMembership reads this account's own membership and follows it (novox/hq ADR 0160): apply is called
|
|
// with every membership the mesh issues this account's node and module, the one already issued first and
|
|
// each later one as it is issued, so a change reaches a running process without a restart. A body that is
|
|
// not a membership is handed to onBad (nil to ignore it) and skipped.
|
|
//
|
|
// It reads the membership from the grants part's state memberships, its own key (novox/hq ADR 0571 clause 7),
|
|
// watched so the value there now comes first and each change after it; and it hears the subject the grants part
|
|
// publishes on until the lease write. The assignments stream is no longer read: the bus module's tools delete it
|
|
// at the lease write (research 072 row 33). A state not there yet, or not readable, is told to onBad and the
|
|
// subject is followed alone. The returned error is a refusal to follow at all.
|
|
func (c *Conn) FollowMembership(apply func(membership.Membership), onBad func(error)) error {
|
|
if c.account.Node == "" || c.account.Module == "" {
|
|
return fmt.Errorf("the account names no node or module, so it has no membership")
|
|
}
|
|
bad := func(err error) {
|
|
if onBad != nil {
|
|
onBad(err)
|
|
}
|
|
}
|
|
take := func(body []byte) {
|
|
issued, err := membership.DecodeGrantsMembership(body)
|
|
if err != nil {
|
|
bad(fmt.Errorf("a membership arrived that is not one: %w", err))
|
|
return
|
|
}
|
|
apply(issued)
|
|
}
|
|
subject := membership.Subject(c.account.Node, c.account.Module)
|
|
if _, err := c.nc.Subscribe(subject, func(m *nats.Msg) { take(m.Data) }); err != nil {
|
|
return fmt.Errorf("cannot follow %s: %w", subject, err)
|
|
}
|
|
key := membership.GrantsKey(c.account.Node, c.account.Module)
|
|
js, err := jetstream.New(c.nc)
|
|
if err != nil {
|
|
bad(fmt.Errorf("cannot read %s of %s (%v); following %s alone", key, membership.GrantsBucket, err, subject))
|
|
return nil
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
kv, err := js.KeyValue(ctx, membership.GrantsBucket)
|
|
if err != nil {
|
|
bad(fmt.Errorf("cannot read %s of %s yet (%v); following %s alone", key, membership.GrantsBucket, err, subject))
|
|
return nil
|
|
}
|
|
w, err := kv.Watch(context.Background(), key)
|
|
if err != nil {
|
|
bad(fmt.Errorf("cannot watch %s of %s (%v); following %s alone", key, membership.GrantsBucket, err, subject))
|
|
return nil
|
|
}
|
|
go func() {
|
|
for e := range w.Updates() {
|
|
if e != nil && e.Operation() == jetstream.KeyValuePut {
|
|
take(e.Value())
|
|
}
|
|
}
|
|
}()
|
|
return nil
|
|
}
|