73 lines
2.0 KiB
Go
73 lines
2.0 KiB
Go
package outbox
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
|
|
"github.com/tech/sendico/pkg/merrors"
|
|
pmessaging "github.com/tech/sendico/pkg/messaging"
|
|
me "github.com/tech/sendico/pkg/messaging/envelope"
|
|
pmessagingreliable "github.com/tech/sendico/pkg/messaging/reliable"
|
|
"github.com/tech/sendico/pkg/mlogger"
|
|
cfgmodel "github.com/tech/sendico/pkg/model"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
// ReliableRuntime owns a reliable producer lifecycle for gateway outbox dispatch.
|
|
type ReliableRuntime struct {
|
|
once sync.Once
|
|
cancel context.CancelFunc
|
|
producer *pmessagingreliable.ReliableProducer
|
|
settings pmessagingreliable.Settings
|
|
initErr error
|
|
}
|
|
|
|
func (r *ReliableRuntime) Start(logger mlogger.Logger, direct pmessaging.Producer, store Store, messagingSettings cfgmodel.SettingsT, opts ...pmessagingreliable.Option) error {
|
|
if r == nil {
|
|
return nil
|
|
}
|
|
if logger == nil {
|
|
logger = zap.NewNop()
|
|
}
|
|
logger = logger.Named("outbox_reliable")
|
|
|
|
r.once.Do(func() {
|
|
reliableProducer, settings, err := NewReliableProducer(logger, direct, store, messagingSettings, opts...)
|
|
if err != nil {
|
|
r.initErr = err
|
|
return
|
|
}
|
|
r.producer = reliableProducer
|
|
r.settings = settings
|
|
if r.producer == nil || direct == nil {
|
|
logger.Info("Outbox reliable publisher disabled", zap.Bool("enabled", settings.Enabled))
|
|
return
|
|
}
|
|
logger.Info("Outbox reliable publisher configured",
|
|
zap.Bool("enabled", settings.Enabled),
|
|
zap.Int("batch_size", settings.BatchSize),
|
|
zap.Int("poll_interval_seconds", settings.PollIntervalSeconds),
|
|
zap.Int("max_attempts", settings.MaxAttempts))
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
r.cancel = cancel
|
|
go r.producer.Run(ctx)
|
|
})
|
|
|
|
return r.initErr
|
|
}
|
|
|
|
func (r *ReliableRuntime) Send(ctx context.Context, envelope me.Envelope) error {
|
|
if r == nil || r.producer == nil {
|
|
return merrors.Internal("reliable outbox producer is not configured")
|
|
}
|
|
return r.producer.SendWithOutbox(ctx, envelope)
|
|
}
|
|
|
|
func (r *ReliableRuntime) Stop() {
|
|
if r == nil || r.cancel == nil {
|
|
return
|
|
}
|
|
r.cancel()
|
|
}
|