mats
16655e8dbc
- deploy/k8s/25-tracking-orders.yaml: SHELL-REBUILD gating removed (LIVE) — replicas 3 + RollingUpdate, actor-pool=tracking label + POD_IP/NAMESPACE, REDIS_ONLY=1, TRACKING_PARTITIONS=16, HPA 3..12 - deploy/k8s/00-namespace-config.yaml: tracking PVC note → rebuild live (mount vestigial, kept for rollback) - deploy/docker-compose.yml: TRACKING_PARTITIONS=16 on the reader producer + tracking consumer (partitioned ingest in the dev stack too) - deploy/k8s/build-push.sh: gocartactor named context for the tracking image - platform: rabbit.PublishEventTo (caller routing key) + event.PartitionOf / PartitionRoutingKey (shared §8 fnv1a hash) + tests - docs: plan-tracking-shell-rebuild.md (config write half, pod-watcher assignment, real-AOF tape gate, R6 re-measure), status.md, CONTEXT, event-sourcing-stance, platform-architecture-review, read-model-projections; scripts/load-test-coordinator.sh - submodules: slask-tracking ecbf894 (shell rebuild), slask-finder 84f7ff7 (§8 partition-routed producer) 🤖 Generated with Codebuff Co-Authored-By: Codebuff <noreply@codebuff.com>
74 lines
2.2 KiB
Go
74 lines
2.2 KiB
Go
package rabbit
|
|
|
|
import (
|
|
"encoding/json"
|
|
"log"
|
|
"time"
|
|
|
|
"git.k6n.net/mats/platform/event"
|
|
amqp "github.com/rabbitmq/amqp091-go"
|
|
)
|
|
|
|
// PublishEvent marshals an event.Event and publishes it to exchange with
|
|
// routingKey = string(ev.Type). Opens and closes a channel per call. The
|
|
// exchange must already exist (declare it with ListenToPattern or call
|
|
// ExchangeDeclare explicitly before the first publish).
|
|
func PublishEvent(c *amqp.Connection, exchange string, ev event.Event) error {
|
|
return PublishEventTo(c, exchange, string(ev.Type), ev)
|
|
}
|
|
|
|
// PublishEventTo marshals an event.Event and publishes it to exchange with the
|
|
// caller-supplied routingKey. PublishEvent hardcodes routingKey =
|
|
// string(ev.Type); partitioned producers (track.partition.{N}) supply their own
|
|
// routing key here. Opens and closes a channel per call; the exchange must
|
|
// already exist.
|
|
func PublishEventTo(c *amqp.Connection, exchange, routingKey string, ev event.Event) error {
|
|
body, err := json.Marshal(ev)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ch, err := c.Channel()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer ch.Close()
|
|
return ch.Publish(exchange, routingKey, true, false, amqp.Publishing{
|
|
ContentType: "application/json",
|
|
Body: body,
|
|
DeliveryMode: amqp.Persistent,
|
|
Timestamp: time.Now(),
|
|
})
|
|
}
|
|
|
|
// ListenToPattern binds an exclusive queue with a routing-key pattern (wildcard)
|
|
// on a durable topic exchange, so one subscription receives every message
|
|
// matching the pattern. A handler error nacks (drops) that single message and
|
|
// keeps consuming.
|
|
func ListenToPattern(ch *amqp.Channel, exchange, pattern string, handler func(amqp.Delivery) error) error {
|
|
if err := ch.ExchangeDeclare(exchange, "topic", true, false, false, false, nil); err != nil {
|
|
return err
|
|
}
|
|
q, err := ch.QueueDeclare("", false, false, true, false, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := ch.QueueBind(q.Name, pattern, exchange, false, nil); err != nil {
|
|
return err
|
|
}
|
|
msgs, err := ch.Consume(q.Name, "", false, false, false, false, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
go func() {
|
|
for d := range msgs {
|
|
if err := handler(d); err != nil {
|
|
log.Printf("Error processing pattern message: %v", err)
|
|
d.Nack(false, false)
|
|
} else {
|
|
d.Ack(false)
|
|
}
|
|
}
|
|
}()
|
|
return nil
|
|
}
|