70 lines
2.3 KiB
Go
70 lines
2.3 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"mesh-grants/internal/grants"
|
|
)
|
|
|
|
// creating creates a stream a manifest declares that the bus lacks (issue 1790; the operator, 2026-10-26 08:54): the
|
|
// create alone, which the account is granted by the stream's name, never an update or a delete. The one place the part
|
|
// creates on the bus.
|
|
type creating struct {
|
|
ctx context.Context
|
|
js jetstream.JetStream
|
|
}
|
|
|
|
// CreateConsumer creates one durable consumer an account needs, by its stream and name (ADR 0561 §6): the create alone;
|
|
// one the bus holds answers ErrHeld and is left as it is.
|
|
func (c creating) CreateConsumer(stream string, cfg jetstream.ConsumerConfig) error {
|
|
ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
|
|
defer cancel()
|
|
if _, err := c.js.Consumer(ctx, stream, cfg.Durable); err == nil {
|
|
return grants.ErrHeld
|
|
}
|
|
_, err := c.js.CreateConsumer(ctx, stream, cfg)
|
|
switch {
|
|
case err == nil:
|
|
return nil
|
|
case errors.Is(err, jetstream.ErrConsumerExists):
|
|
return grants.ErrHeld
|
|
case errors.Is(err, context.DeadlineExceeded):
|
|
return fmt.Errorf("no answer in %s: a create the bus refuses is answered by none (%w)", requestWithin, err)
|
|
}
|
|
return err
|
|
}
|
|
|
|
// MatchConsumer brings a held consumer to cfg on a person's ask (grants_consumer_match), keeping where it delivers from:
|
|
// the server refuses a changed deliver policy, and a consumer moved would replay or lose what it holds.
|
|
func (c creating) MatchConsumer(stream string, cfg jetstream.ConsumerConfig) error {
|
|
ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
|
|
defer cancel()
|
|
held, err := c.js.Consumer(ctx, stream, cfg.Durable)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
have := held.CachedInfo().Config
|
|
cfg.DeliverPolicy, cfg.OptStartSeq, cfg.OptStartTime = have.DeliverPolicy, have.OptStartSeq, have.OptStartTime
|
|
_, err = c.js.UpdateConsumer(ctx, stream, cfg)
|
|
return err
|
|
}
|
|
|
|
func (c creating) CreateDeclared(cfg jetstream.StreamConfig) error {
|
|
ctx, cancel := context.WithTimeout(c.ctx, requestWithin)
|
|
defer cancel()
|
|
_, err := c.js.CreateStream(ctx, cfg)
|
|
switch {
|
|
case err == nil:
|
|
return nil
|
|
case errors.Is(err, jetstream.ErrStreamNameAlreadyInUse):
|
|
return grants.ErrHeld
|
|
case errors.Is(err, context.DeadlineExceeded):
|
|
return fmt.Errorf("no answer in %s: a create the bus refuses is answered by none (%w)", requestWithin, err)
|
|
}
|
|
return err
|
|
}
|