diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go index 25e4b970..3c6860fc 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go @@ -195,23 +195,23 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro go onReconnect(ctx, again, said, logger) // What a merge read of each module directory, by its tree (issue 1126): a merge reads only the directories it changed. inputs := newKept(process.OptionalBucket(ctx, js, inputsBucket, "a merge's module directories are kept in memory only, so the first merge after a start reads every one", logger.Printf), logger) - cons, err := cred.EventConsumer(ctx, js, eventsStream) - if err != nil { - // Essential: the forge's events are what the bridge carries. - return fmt.Errorf("the consumer %s_%s on %s cannot be read (%v): the bridge hears no forge event without it, so it stops; where the bus refused it or did not answer, this module's consumes (made by the controller with its grant) have not reached the bus yet", cred.Node, cred.Module, eventsStream, err) - } - cc, err := process.HearEvents(cons, func(m jetstream.Msg) { + // The module's one pull of its durable consumer on the shared events stream is its event hub; the bridge has the one + // handler, for every event its consumes filter lets through. Essential: the forge's events are what the bridge carries. + hub := process.NewEventHub(logger.Printf) + if err := hub.Handle("mesh.mod.*.event.>", func(m jetstream.Msg) { hear(&turn, func() { // Bounded: the turn is held, and no sync or other event goes on without it. hctx, cancel := context.WithTimeout(ctx, syncDeadline) defer cancel() handleGrouped(hctx, m, rec, treesVia(hctx, call), filesVia(hctx, call), inputs, groups, logger) }) - }, logger.Printf) - if err != nil { + }); err != nil { return err } - defer cc.Stop() + if err := hub.Start(ctx, js, cred); err != nil { + return fmt.Errorf("%w: the bridge hears no forge event without it, so it stops; where the bus refused it or did not answer, this module's consumes (made by the controller with its grant) have not reached the bus yet", err) + } + defer hub.Stop() select { case <-ctx.Done(): return nil