128 lines
4.6 KiB
Go
128 lines
4.6 KiB
Go
package process
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
)
|
|
|
|
// Message is one piece of work this process's queue consumer delivered, with how many times the bus has
|
|
// delivered it. Nothing acts on its behalf: the handler decides its fate exactly once, by calling Ack,
|
|
// NakWithDelay or Term (mesh-tools node-tools/internal/bus/seat.go's seat work Consume, with the same
|
|
// three outcomes, offered here explicitly instead of by a handler's return value, since the caller
|
|
// chooses its own delay before a message comes back).
|
|
type Message struct {
|
|
Subject string
|
|
Body []byte
|
|
Headers map[string]string
|
|
// Delivered is how many times the bus has handed this message over, this time included.
|
|
Delivered uint64
|
|
|
|
ack func() error
|
|
nak func(time.Duration) error
|
|
term func() error
|
|
}
|
|
|
|
// Ack acknowledges the message: the bus never delivers it again.
|
|
func (m *Message) Ack() error { return m.ack() }
|
|
|
|
// NakWithDelay asks for the message to be delivered again no sooner than delay, its delivery count
|
|
// raised by one.
|
|
func (m *Message) NakWithDelay(delay time.Duration) error { return m.nak(delay) }
|
|
|
|
// Term terminates the message: the bus never delivers it again, and does not count it a dead letter the
|
|
// way running out of deliveries does.
|
|
func (m *Message) Term() error { return m.term() }
|
|
|
|
// ProcessID is this account's own process id, the last segment of its user name <node>.<module>.<id>
|
|
// (mesh-controller internal/broker, Principal.Username for KindProcess).
|
|
func (a Account) ProcessID() string {
|
|
prefix := a.Node + "." + a.Module + "."
|
|
if !strings.HasPrefix(a.User, prefix) {
|
|
return ""
|
|
}
|
|
return strings.TrimPrefix(a.User, prefix)
|
|
}
|
|
|
|
// QueueConsumerName is the one durable consumer a process's bus.queues names it on its work-queue stream
|
|
// (novox/hq ADR 0436 clauses 1, 2, 6; mesh-controller internal/broker/modstreams.go QueueConsumer):
|
|
// <node>_<module>_<process id>.
|
|
func (a Account) QueueConsumerName() string { return a.Node + "_" + a.Module + "_" + a.ProcessID() }
|
|
|
|
// errNoMessage is one pull that fetched nothing within its wait: not an error to the caller, just a
|
|
// reason to pull again.
|
|
var errNoMessage = errors.New("no message this pull")
|
|
|
|
// Consume binds this process's one durable consumer on stream (QueueConsumerName) and pulls from it, one
|
|
// message at a time, handing each to handle until ctx ends: it then stops pulling and returns ctx.Err(),
|
|
// never acknowledging a message still in handle's hands — only handle decides a message's fate, by
|
|
// calling Ack, NakWithDelay or Term on it.
|
|
func (c *Conn) Consume(ctx context.Context, stream string, handle func(*Message)) error {
|
|
js, err := jetstream.New(c.nc)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
name := c.account.QueueConsumerName()
|
|
cons, err := js.Consumer(ctx, stream, name)
|
|
if err != nil {
|
|
return fmt.Errorf("binding %s's queue consumer %s on %s: %w", c.account.Module, name, stream, err)
|
|
}
|
|
return runQueue(ctx, fetchOne(cons), handle)
|
|
}
|
|
|
|
// fetchOne is one pull of at most one message, non-blocking past ctx: Fetch itself does not take a
|
|
// context, so this selects on its message channel and ctx together, and returns promptly once ctx ends
|
|
// even while a fetch is still outstanding.
|
|
func fetchOne(cons jetstream.Consumer) func(ctx context.Context) (*Message, error) {
|
|
return func(ctx context.Context) (*Message, error) {
|
|
batch, err := cons.Fetch(1, jetstream.FetchMaxWait(5*time.Second))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
select {
|
|
case msg, ok := <-batch.Messages():
|
|
if !ok {
|
|
return nil, errNoMessage
|
|
}
|
|
headers := map[string]string{}
|
|
for k := range msg.Headers() {
|
|
headers[k] = msg.Headers().Get(k)
|
|
}
|
|
var delivered uint64 = 1
|
|
if meta, err := msg.Metadata(); err == nil && meta != nil {
|
|
delivered = meta.NumDelivered
|
|
}
|
|
return &Message{Subject: msg.Subject(), Body: append([]byte(nil), msg.Data()...), Headers: headers,
|
|
Delivered: delivered, ack: msg.Ack, nak: msg.NakWithDelay, term: msg.Term}, nil
|
|
case <-ctx.Done():
|
|
return nil, ctx.Err()
|
|
}
|
|
}
|
|
}
|
|
|
|
// runQueue is Consume's loop, apart from the bus so a unit test can drive it with a fake next
|
|
// (novox/hq issue 1190 AC4): it checks ctx before every pull, and stops as soon as next or the check
|
|
// says ctx has ended, without touching the message past handing it to handle.
|
|
func runQueue(ctx context.Context, next func(context.Context) (*Message, error), handle func(*Message)) error {
|
|
for {
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
msg, err := next(ctx)
|
|
if err != nil {
|
|
if ctx.Err() != nil {
|
|
return ctx.Err()
|
|
}
|
|
if errors.Is(err, errNoMessage) {
|
|
continue
|
|
}
|
|
return err
|
|
}
|
|
handle(msg)
|
|
}
|
|
}
|