R-3: W1.12: one stream name
A seat event's stream has one bus name, seatdef.KeptStream's, which the generated contract carries as Stream<E>: HearSeatEvents takes it as it stands and refuses a dotted or unprefixed name by name, so no part makes the name a second time (SEAT_SEAT_...). Builds: design 47 G §260 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
@@ -11,6 +11,7 @@ import (
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatdef"
|
||||
"git.novox.be/novox/mesh-sdk/go/testbus"
|
||||
)
|
||||
|
||||
@@ -80,8 +81,8 @@ func TestAModuleHearsASeatEventKeptOnItsOwnStream(t *testing.T) {
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
kept, subjects := "node-engine.running", []string{"mesh.seat.node-engine.event.running.>"}
|
||||
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: process.SeatStreamName(kept), Subjects: subjects, MaxMsgsPerSubject: 1}); err != nil {
|
||||
kept, subjects := seatdef.KeptStream("node-engine", seatdef.Message{Name: "running"}), []string{"mesh.seat.node-engine.event.running.>"}
|
||||
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: kept, Subjects: subjects, MaxMsgsPerSubject: 1}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, m := range []struct{ subject, fp string }{
|
||||
@@ -99,7 +100,7 @@ func TestAModuleHearsASeatEventKeptOnItsOwnStream(t *testing.T) {
|
||||
if _, err := cred.HearSeatEvents(ctx, js, kept, hear, t.Logf); err == nil {
|
||||
t.Fatal("no consumer was made, and hearing said nothing")
|
||||
}
|
||||
if _, err := js.CreateOrUpdateConsumer(ctx, process.SeatStreamName(kept), cred.SeatEventConsumerConfig(subjects, true)); err != nil {
|
||||
if _, err := js.CreateOrUpdateConsumer(ctx, kept, cred.SeatEventConsumerConfig(subjects, true)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cc, err := cred.HearSeatEvents(ctx, js, kept, hear, t.Logf)
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
package process
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
@@ -36,10 +38,15 @@ func TestTheHubMatchesSubjectsAndRefusesOverlappingPatterns(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// W1.12: a seat's event kept in a stream of its own is heard on one bus name and one durable, the maker's and the
|
||||
// hearer's alike; an event kept last per subject is delivered from the last of each subject.
|
||||
// hearer's alike; an event kept last per subject is delivered from the last of each subject. The name is the
|
||||
// contract's Stream<E> as it stands (seatdef.KeptStream), never made again here: a dotted or unprefixed name is
|
||||
// refused by name before the bus is asked.
|
||||
a := Account{Node: "ace", Module: "mesh-grants"}
|
||||
if got := SeatStreamName("node-engine.running"); got != "SEAT_node-engine_running" {
|
||||
t.Errorf("the bus's name of node-engine.running is %q", got)
|
||||
for _, stream := range []string{"node-engine.running", "SEAT_node-engine.running", "SEAT_", "EVENTS"} {
|
||||
if _, err := a.HearSeatEvents(context.Background(), nil, stream, func(jetstream.Msg) {}, nil); err == nil ||
|
||||
!strings.Contains(err.Error(), "is not the bus's name of a seat event's stream") {
|
||||
t.Errorf("hearing %q: %v", stream, err)
|
||||
}
|
||||
}
|
||||
cfg := a.SeatEventConsumerConfig([]string{"mesh.seat.node-engine.event.running.>"}, true)
|
||||
if cfg.Durable != "ace_mesh-grants" || cfg.DeliverPolicy != jetstream.DeliverLastPerSubjectPolicy ||
|
||||
|
||||
+20
-17
@@ -6,21 +6,21 @@ import (
|
||||
"strings"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/seatdef"
|
||||
)
|
||||
|
||||
// A seat's event kept in a stream of its own (seatdef's `keep`: the last message of each subject, or a stream with an age
|
||||
// and a cap), not on the shared events stream (novox/hq design 47: the node-engine seat's `running.<node>.<module>`, the
|
||||
// seat mesh-delivery's `owed`). A generated contract names that stream as `<seat>.<event>` (Stream<E>, seatdef.KeptStream);
|
||||
// these are the one rule, for the part that makes the stream and its consumers (the grants part, from each module's
|
||||
// `consumes`) and for every part that hears it, of the stream's name on the bus and the module's durable on it.
|
||||
// seat mesh-delivery's `owed`). Its name on the bus is seatdef.KeptStream's, `SEAT_<seat>_<event>`, which a generated
|
||||
// contract carries as Stream<E>: one name in one place. These are the one rule, for the part that makes the stream's
|
||||
// consumers (the grants part, from each module's `consumes`) and for every part that hears it, of the module's durable
|
||||
// on it.
|
||||
|
||||
// SeatStreamPrefix starts the bus's name of every stream a seat's event is kept in apart from its family.
|
||||
const SeatStreamPrefix = "SEAT_"
|
||||
|
||||
// SeatStreamName is the bus's name of the stream a contract names kept (`<seat>.<event>`): the bus takes no dot in a
|
||||
// stream's name, so `SEAT_<seat>_<event>`, each dot an underscore (`node-engine.running` is `SEAT_node-engine_running`).
|
||||
func SeatStreamName(kept string) string {
|
||||
return SeatStreamPrefix + strings.ReplaceAll(kept, ".", "_")
|
||||
// keptStreamName is whether stream is a name seatdef.KeptStream gives: the bus's, with its prefix and no dot.
|
||||
func keptStreamName(stream string) bool {
|
||||
return strings.HasPrefix(stream, seatdef.KeptStreamPrefix) && len(stream) > len(seatdef.KeptStreamPrefix) &&
|
||||
!strings.Contains(stream, ".")
|
||||
}
|
||||
|
||||
// SeatEventConsumerConfig is the module's durable on a seat event's own stream as its maker makes it: <node>_<module>, as
|
||||
@@ -35,26 +35,29 @@ func (a Account) SeatEventConsumerConfig(subjects []string, lastPerSubject bool)
|
||||
return cfg
|
||||
}
|
||||
|
||||
// HearSeatEvents binds the module's durable on the stream of kept (a contract's Stream<E>) and pulls it, handing each
|
||||
// HearSeatEvents binds the module's durable on the stream (a contract's Stream<E>, as it stands) and pulls it, handing each
|
||||
// message to handle, which acknowledges it; every error the bus reports while hearing is said through logf. Like
|
||||
// EventHub it makes neither the stream nor the consumer: one the bus does not hold yet is this call's error, naming
|
||||
// both, and a later call binds it. The caller stops the returned context.
|
||||
func (a Account) HearSeatEvents(ctx context.Context, js jetstream.JetStream, kept string, handle func(jetstream.Msg), logf func(string, ...any)) (jetstream.ConsumeContext, error) {
|
||||
func (a Account) HearSeatEvents(ctx context.Context, js jetstream.JetStream, stream string, handle func(jetstream.Msg), logf func(string, ...any)) (jetstream.ConsumeContext, error) {
|
||||
if !keptStreamName(stream) {
|
||||
return nil, fmt.Errorf("%q is not the bus's name of a seat event's stream: a contract's Stream<E> is %s<seat>_<event>, "+
|
||||
"with no dot (seatdef.KeptStream)", stream, seatdef.KeptStreamPrefix)
|
||||
}
|
||||
if handle == nil {
|
||||
return nil, fmt.Errorf("hearing %s needs a handler", kept)
|
||||
return nil, fmt.Errorf("hearing %s needs a handler", stream)
|
||||
}
|
||||
if logf == nil {
|
||||
logf = func(string, ...any) {}
|
||||
}
|
||||
stream := SeatStreamName(kept)
|
||||
cons, err := a.EventConsumer(ctx, js, stream)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the consumer %s on %s (the seat's stream %s) is not there: %w", a.EventConsumerName(), stream, kept, err)
|
||||
return nil, fmt.Errorf("the consumer %s on %s is not there: %w", a.EventConsumerName(), stream, err)
|
||||
}
|
||||
cc, err := HearEvents(cons, handle, logf)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the consumer %s on %s (the seat's stream %s) is not heard: %w", a.EventConsumerName(), stream, kept, err)
|
||||
return nil, fmt.Errorf("the consumer %s on %s is not heard: %w", a.EventConsumerName(), stream, err)
|
||||
}
|
||||
logf("events: the consumer %s on %s (the seat's stream %s) is heard", a.EventConsumerName(), stream, kept)
|
||||
logf("events: the consumer %s on %s is heard", a.EventConsumerName(), stream)
|
||||
return cc, nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user