141 lines
3.4 KiB
Go
141 lines
3.4 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"sync"
|
|
"time"
|
|
|
|
amqp "github.com/rabbitmq/amqp091-go"
|
|
)
|
|
|
|
// rabbitPublisher is the broker-delivery side used by the outbox relay. It dials
|
|
// lazily and reconnects on failure, so the relay can be started even when the
|
|
// broker is down at boot: messages accumulate durably in the outbox and drain
|
|
// once the broker is reachable.
|
|
type rabbitPublisher struct {
|
|
url string
|
|
mu sync.Mutex
|
|
conn *amqp.Connection
|
|
ch *amqp.Channel
|
|
}
|
|
|
|
func newRabbitPublisher(url string) *rabbitPublisher {
|
|
return &rabbitPublisher{url: url}
|
|
}
|
|
|
|
func (p *rabbitPublisher) connect() error {
|
|
conn, err := amqp.Dial(p.url)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ch, err := conn.Channel()
|
|
if err != nil {
|
|
_ = conn.Close()
|
|
return err
|
|
}
|
|
p.conn, p.ch = conn, ch
|
|
return nil
|
|
}
|
|
|
|
func (p *rabbitPublisher) reset() {
|
|
if p.ch != nil {
|
|
_ = p.ch.Close()
|
|
}
|
|
if p.conn != nil {
|
|
_ = p.conn.Close()
|
|
}
|
|
p.ch, p.conn = nil, nil
|
|
}
|
|
|
|
func (p *rabbitPublisher) Publish(exchange, routingKey string, body []byte) error {
|
|
p.mu.Lock()
|
|
defer p.mu.Unlock()
|
|
if p.ch == nil {
|
|
if err := p.connect(); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
// Bound the publish so a half-open/black-holed connection can't wedge the
|
|
// outbox relay indefinitely; on timeout the error path resets and the relay
|
|
// retries on its next tick.
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
err := p.ch.PublishWithContext(ctx, exchange, routingKey, false, false,
|
|
amqp.Publishing{ContentType: "application/json", Body: body})
|
|
if err != nil {
|
|
p.reset() // force a reconnect on the next attempt
|
|
}
|
|
return err
|
|
}
|
|
|
|
// ingestOrderFromQueue records one order from the queue. It uses the exact same
|
|
// idempotent business logic as the HTTP endpoint.
|
|
func ingestOrderFromQueue(ctx context.Context, s *server, body []byte) error {
|
|
var req fromCheckoutReq
|
|
if err := json.Unmarshal(body, &req); err != nil {
|
|
return err
|
|
}
|
|
if len(req.Lines) == 0 {
|
|
return nil // nothing actionable
|
|
}
|
|
|
|
_, _, _, _, runErr, err := s.createOrderFromCheckoutInternal(ctx, &req)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if runErr != nil {
|
|
return runErr
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// startOrderIngest connects to AMQP and consumes "order-queue" until ctx is
|
|
// done.
|
|
func startOrderIngest(ctx context.Context, amqpURL string, s *server) {
|
|
conn, err := amqp.Dial(amqpURL)
|
|
if err != nil {
|
|
s.logger.Warn("order ingest disabled: amqp dial failed", "err", err)
|
|
return
|
|
}
|
|
ch, err := conn.Channel()
|
|
if err != nil {
|
|
s.logger.Warn("order ingest disabled: channel failed", "err", err)
|
|
_ = conn.Close()
|
|
return
|
|
}
|
|
q, err := ch.QueueDeclare("order-queue", false, false, false, false, nil)
|
|
if err != nil {
|
|
s.logger.Warn("order ingest disabled: queue declare failed", "err", err)
|
|
_ = ch.Close()
|
|
_ = conn.Close()
|
|
return
|
|
}
|
|
msgs, err := ch.Consume(q.Name, "order-service", true, false, false, false, nil)
|
|
if err != nil {
|
|
s.logger.Warn("order ingest disabled: consume failed", "err", err)
|
|
_ = ch.Close()
|
|
_ = conn.Close()
|
|
return
|
|
}
|
|
s.logger.Info("ingesting orders from order-queue")
|
|
go func() {
|
|
defer conn.Close()
|
|
defer ch.Close()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case d, ok := <-msgs:
|
|
if !ok {
|
|
s.logger.Warn("order ingest: channel closed")
|
|
return
|
|
}
|
|
if err := ingestOrderFromQueue(ctx, s, d.Body); err != nil {
|
|
s.logger.Error("order queue ingest failed", "err", err)
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
}
|