103 lines
3.2 KiB
Go
103 lines
3.2 KiB
Go
package buildstep
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
)
|
|
|
|
// JetStreamAsks is a build seat's workers on the bus: the shared pull consumer of each kind on the seat's stream
|
|
// (delivery.AskStream, AskWorker), which the broker makes when the seat is declared; a holder only binds them.
|
|
type JetStreamAsks struct {
|
|
JS jetstream.JetStream
|
|
Language string
|
|
|
|
mu sync.Mutex
|
|
cons map[string]jetstream.Consumer
|
|
}
|
|
|
|
func (s *JetStreamAsks) consumer(ctx context.Context, kind string) (jetstream.Consumer, error) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if c, ok := s.cons[kind]; ok {
|
|
return c, nil
|
|
}
|
|
stream, err := delivery.AskStream(s.Language)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
worker, err := delivery.AskWorker(s.Language, kind)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
c, err := s.JS.Consumer(ctx, stream, worker)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("binding the %s worker %s on %s: %w", kind, worker, stream, err)
|
|
}
|
|
if s.cons == nil {
|
|
s.cons = map[string]jetstream.Consumer{}
|
|
}
|
|
s.cons[kind] = c
|
|
return c, nil
|
|
}
|
|
|
|
// Next is AskSource's: a fetch of one with no wait when wait is 0, else one that waits at most wait.
|
|
func (s *JetStreamAsks) Next(ctx context.Context, kind string, wait time.Duration) (*Pulled, error) {
|
|
c, err := s.consumer(ctx, kind)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var batch jetstream.MessageBatch
|
|
if wait <= 0 {
|
|
batch, err = c.FetchNoWait(1)
|
|
} else {
|
|
batch, err = c.Fetch(1, jetstream.FetchMaxWait(wait))
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
select {
|
|
case msg, ok := <-batch.Messages():
|
|
if !ok {
|
|
return nil, batch.Error()
|
|
}
|
|
var delivered uint64 = 1
|
|
if meta, err := msg.Metadata(); err == nil && meta != nil {
|
|
delivered = meta.NumDelivered
|
|
}
|
|
return &Pulled{Data: append([]byte(nil), msg.Data()...), Delivered: delivered,
|
|
Ack: msg.Ack, InProgress: msg.InProgress, Nak: msg.Nak, Term: msg.Term}, nil
|
|
case <-ctx.Done():
|
|
return nil, ctx.Err()
|
|
}
|
|
}
|
|
|
|
// DeclareAskWorkers makes language's seat stream and its three workers on a bus with no broker: a test's (the build
|
|
// modules' integration tests start one). On the mesh the broker declares them from the seat, as it declares every
|
|
// seat's stream (mesh-controller internal/broker SeatStreams); a holder never does.
|
|
func DeclareAskWorkers(ctx context.Context, js jetstream.JetStream, language string, ackWait time.Duration) error {
|
|
stream, err := delivery.AskStream(language)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
seat, _ := delivery.BuildSeat(language)
|
|
if _, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{"mesh.seat." + seat + ".accept.>"},
|
|
Retention: jetstream.WorkQueuePolicy}); err != nil {
|
|
return fmt.Errorf("the stream %s: %w", stream, err)
|
|
}
|
|
for _, kind := range delivery.AskKinds {
|
|
worker, _ := delivery.AskWorker(language, kind)
|
|
subject, _ := delivery.AskSubject(language, kind)
|
|
if _, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{Durable: worker, FilterSubject: subject,
|
|
AckPolicy: jetstream.AckExplicitPolicy, AckWait: ackWait, MaxDeliver: 5}); err != nil {
|
|
return fmt.Errorf("the worker %s: %w", worker, err)
|
|
}
|
|
}
|
|
return nil
|
|
}
|