Documentation
¶
Overview ¶
Package nats provides the NATS JetStream connection, subject routing, and event/queue subscribers shared by all Tupic services.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func NewConnection ¶
Types ¶
type Config ¶
type Config struct {
URL string `mapstructure:"url"`
Token string `mapstructure:"token"`
SubjectPrefix string `mapstructure:"subject_prefix"`
AppSlug string `mapstructure:"-"`
}
Config carries NATS connection settings plus the service identity used for durable consumer naming and connection naming. The mapstructure tags let a service load it directly from JSON (alias it); AppSlug is injected by the service from its identity constant, not loaded, so it is skipped.
type EventSubscriber ¶
type EventSubscriber struct {
// contains filtered or unexported fields
}
EventSubscriber handles non-queue subjects (integration events from other services).
func NewEventSubscriber ¶
func NewEventSubscriber( l logger.Logger, js natslib.JetStreamContext, handler MessageRouter, cfg Config, ) *EventSubscriber
func (*EventSubscriber) Stop ¶
func (s *EventSubscriber) Stop(ctx context.Context) error
Stop cancels the handler context, unsubscribes all consumers (collecting any errors), then waits for in-flight handlers to drain. All subscriptions are unsubscribed regardless of individual errors. If ctx expires before the drain completes a warning is logged and the function returns so downstream lifecycle hooks are not blocked.
type FailedMessage ¶
type FailedMessage struct {
ID uuid.UUID `gorm:"column:id;type:char(36);primaryKey"`
Type string `gorm:"column:type;type:varchar(255);not null"`
Version string `gorm:"column:version;type:varchar(20);not null"`
Payload datatypes.JSON `gorm:"column:payload;type:jsonb;not null"`
Attempts int `gorm:"column:attempts;type:int;not null"`
LastError string `gorm:"column:last_error;type:text;not null"`
FailedAt time.Time `gorm:"column:failed_at;type:timestamp;not null"`
}
FailedMessage is the GORM model for the failed_tasks DLQ table. It records every message that the Worker gave up on after MaxDeliver attempts or a terminal error from the handler.
The table name remains "failed_tasks" for backward compatibility with existing migrations. Operators query / re-publish from here; nothing inside the running app reads it on a hot path.
func (*FailedMessage) TableName ¶
func (*FailedMessage) TableName() string
type MessageRouter ¶
type MessageRouter interface {
Subjects() []string
Handle(ctx context.Context, subject string, m messaging2.Message) error
}
MessageRouter is the subject→handler dispatch the NATS subscribers depend on. They declare the narrow behaviour they need rather than importing the concrete router, so the composition root owns the binding (the messaging.Router satisfies it).
type QueueSubscriber ¶
type QueueSubscriber struct {
// contains filtered or unexported fields
}
QueueSubscriber subscribes to queues.* subjects and dispatches each task to the Router. Terminal failures (apperror) and exhausted MaxDeliver are written to the failed_messages DLQ and Term'd; transient failures are Nak'd for JetStream retry.
func NewQueueSubscriber ¶
func NewQueueSubscriber( l logger.Logger, c clock.Clock, js natslib.JetStreamContext, router MessageRouter, cfg Config, db *gorm.DB, ) (*QueueSubscriber, error)
func (*QueueSubscriber) Start ¶
func (w *QueueSubscriber) Start(_ context.Context) error
Start subscribes to every registered queues.* subject. Each subject gets its own durable consumer so JetStream load-balances across replicas naturally.
func (*QueueSubscriber) Stop ¶
func (w *QueueSubscriber) Stop(ctx context.Context) error
Stop cancels the handler context, unsubscribes all consumers (collecting any errors), then waits for in-flight handlers to drain. All subscriptions are unsubscribed regardless of individual errors. If ctx expires before the drain completes a warning is logged so downstream lifecycle hooks are not blocked.
Source Files