96 lines
4.2 KiB
Go
96 lines
4.2 KiB
Go
package process
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
)
|
|
|
|
// What a module's own principal reaches on the bus beyond a process account's queue (novox/hq issue 1242): its
|
|
// own credential, its own stream and its durable consumer on the shared events stream. These helpers are the one
|
|
// place a long-running part of a module (a daemon the module's delivery runs) learns them, so no part hand-writes
|
|
// the connection, the subject spaces or the consumer's name.
|
|
|
|
// EventsStream is the shared stream every module's events (mesh.mod.<module>.event.>) are on; a module's consumer
|
|
// on it is the durable <node>_<module> the controller makes (mesh-controller internal/broker/nats.go
|
|
// consumerStream, consumerDurable).
|
|
const EventsStream = "EVENTS"
|
|
|
|
// BrokerFileEnv names the environment variable holding the path of the module's own broker credential.
|
|
const BrokerFileEnv = "MESH_BROKER_FILE"
|
|
|
|
// ReadBrokerAccount reads the module's own credential from the file BrokerFileEnv names in getenv.
|
|
func ReadBrokerAccount(getenv func(string) string) (Account, error) {
|
|
path := strings.TrimSpace(getenv(BrokerFileEnv))
|
|
if path == "" {
|
|
return Account{}, errors.New(BrokerFileEnv + " is not set, so there is no credential to hear the bus with")
|
|
}
|
|
return ReadAccount(path)
|
|
}
|
|
|
|
// DialOptions is the option list Dial connects with: named <node>.<module>, as the account's user on its own
|
|
// inbox prefix, reconnecting without limit and retrying a first connect that fails, pinned to the account's
|
|
// fingerprint when it carries one; opts follow, so a caller may add a handler or override one.
|
|
func (a Account) DialOptions(opts ...nats.Option) []nats.Option {
|
|
all := []nats.Option{
|
|
nats.Name(a.Node + "." + a.Module),
|
|
nats.UserInfo(a.User, a.Password),
|
|
nats.CustomInboxPrefix(a.InboxPrefix()),
|
|
nats.MaxReconnects(-1),
|
|
nats.RetryOnFailedConnect(true),
|
|
}
|
|
if strings.TrimSpace(a.Fingerprint) != "" {
|
|
all = append(all, nats.Secure(PinnedToFingerprint(a.Fingerprint)))
|
|
}
|
|
return append(all, opts...)
|
|
}
|
|
|
|
// Dial connects as the module's own principal and opens JetStream on the connection. The caller closes the
|
|
// connection.
|
|
func (a Account) Dial(opts ...nats.Option) (*nats.Conn, jetstream.JetStream, error) {
|
|
nc, err := nats.Connect(a.URL, a.DialOptions(opts...)...)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("connecting to the bus: %w", err)
|
|
}
|
|
js, err := jetstream.New(nc)
|
|
if err != nil {
|
|
nc.Close()
|
|
return nil, nil, err
|
|
}
|
|
return nc, js, nil
|
|
}
|
|
|
|
// EventSubject is where the module's event lands on the shared events stream: mesh.mod.<module>.event.<event>.
|
|
func (a Account) EventSubject(event string) string { return "mesh.mod." + a.Module + ".event." + event }
|
|
|
|
// StreamSubject is where the module's event lands on a stream of its own: mesh.stream.<module>.event.<event>
|
|
// (mesh-controller internal/broker/modstreams.go ModuleStreamSubject).
|
|
func (a Account) StreamSubject(event string) string { return ModuleStreamSubject(a.Module, event) }
|
|
|
|
// ModuleStreamSubject is where a module's event lands on that module's own stream, for a part that names another
|
|
// module's stream (the indexer hears the forge bridge's, the checker's) without holding its credential.
|
|
func ModuleStreamSubject(module, event string) string {
|
|
return "mesh.stream." + module + ".event." + event
|
|
}
|
|
|
|
// EventConsumerName is the module's durable consumer on a stream it is granted: <node>_<module>.
|
|
func (a Account) EventConsumerName() string { return a.Node + "_" + a.Module }
|
|
|
|
// EventConsumer binds the module's durable consumer on stream (EventsStream, or a stream of another module it is
|
|
// granted).
|
|
func (a Account) EventConsumer(ctx context.Context, js jetstream.JetStream, stream string) (jetstream.Consumer, error) {
|
|
return js.Consumer(ctx, stream, a.EventConsumerName())
|
|
}
|
|
|
|
// HearEvents pulls from cons, handing each message to handle, and says every error the bus reports while
|
|
// hearing through logf ("hearing the bus: ..."). The caller stops the returned context.
|
|
func HearEvents(cons jetstream.Consumer, handle func(jetstream.Msg), logf func(string, ...any)) (jetstream.ConsumeContext, error) {
|
|
return cons.Consume(handle, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) {
|
|
logf("hearing the bus: %v", err)
|
|
}))
|
|
}
|