consumer

package
v0.8.1 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: May 29, 2026 License: Apache-2.0 Imports: 21 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type NATSConsumer added in v0.2.0

type NATSConsumer struct {
	// contains filtered or unexported fields
}

NATSConsumer wraps a NATS JetStream consumer for processing test events. Structured run, suite, test, and attempt data is persisted to PostgreSQL, while MongoDB is retained only for the live step buffer used during in-flight execution.

func NewNATSConsumer added in v0.2.0

func NewNATSConsumer(cfg NATSConsumerConfig, logger *slog.Logger, bufferRepo *mongodb.MongoRepository, pgRepo *postgres.PostgresRepository) (*NATSConsumer, error)

NewNATSConsumer creates a new NATS JetStream consumer with Postgres backend and MongoDB-backed live step buffer support.

func (*NATSConsumer) Close added in v0.2.0

func (c *NATSConsumer) Close() error

Close closes the NATS connection

func (*NATSConsumer) Start added in v0.2.0

Start begins consuming messages from NATS

type NATSConsumerConfig added in v0.2.0

type NATSConsumerConfig struct {
	URL          string
	StreamName   string
	ConsumerName string
	BatchSize    int
	MaxWait      time.Duration
	MaxDeliver   int
	AckWait      time.Duration
	DLQSubject   string

	DeferQueueMaxAttempts int
	DeferQueueTTL         time.Duration
	RetainMessages        bool // Deprecated no-op; legacy raw message retention was removed.
}

NATSConsumerConfig holds configuration for NATS consumer

type ReconciliationConfig

type ReconciliationConfig struct {
	// Enabled controls whether reconciliation is active
	Enabled bool

	// MaxBufferSize is the maximum number of events to buffer per run
	MaxBufferSize int

	// InactivityTTL is the duration after last event before triggering reconciliation
	InactivityTTL time.Duration

	// MaxPasses is the maximum number of reconciliation passes to attempt
	MaxPasses int

	// PassDelay is the delay between reconciliation passes
	PassDelay time.Duration

	// CleanupInterval is how often to check for inactive buffers
	CleanupInterval time.Duration
}

ReconciliationConfig holds configuration for event reconciliation

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL