tech/sendico
2
0
Fork 0
Code Issues 24 Pull Requests 3 Actions Packages Projects Releases Wiki Activity

Fixed messaging buffering issue

Browse Source
This commit is contained in:
Stephan D
2026-01-15 18:52:14 +01:00
parent c775b3fdd4
commit 4786df60ae
40 changed files with 125 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
api/pkg/go.mod
View File

@@ -5,7 +5,7 @@ go 1.24.0
require (
github.com/casbin/casbin/v2 v2.135.0
github.com/casbin/mongodb-adapter/v3 v3.7.0
github.com/go-chi/chi/v5 v5.2.3
github.com/go-chi/chi/v5 v5.2.4
github.com/google/uuid v1.6.0
github.com/mattn/go-colorable v0.1.14
github.com/mitchellh/mapstructure v1.5.0
@@ -93,6 +93,6 @@ require (
golang.org/x/sys v0.40.0 // indirect
golang.org/x/text v0.33.0 // indirect
golang.org/x/time v0.5.0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20260112192933-99fd39fd28a9 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20260114163908-3f89685c29c3 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
)

8
api/pkg/go.sum
View File

@@ -43,8 +43,8 @@ github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4
github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
github.com/go-chi/chi/v5 v5.2.3 h1:WQIt9uxdsAbgIYgid+BpYc+liqQZGMHRaUwp0JUcvdE=
github.com/go-chi/chi/v5 v5.2.3/go.mod h1:L2yAIGWB3H+phAw1NxKwWM+7eUH/lU8pOMm5hHcoops=
github.com/go-chi/chi/v5 v5.2.4 h1:WtFKPHwlywe8Srng8j2BhOD9312j9cGUxG1SP4V2cR4=
github.com/go-chi/chi/v5 v5.2.4/go.mod h1:X7Gx4mteadT3eDOMTsXzmI4/rwUpOwBHLpAfupzFJP0=
github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
@@ -269,8 +269,8 @@ gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk=
gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E=
google.golang.org/genproto/googleapis/api v0.0.0-20251029180050-ab9386a59fda h1:+2XxjfsAu6vqFxwGBRcHiMaDCuZiqXGDUDVWVtrFAnE=
google.golang.org/genproto/googleapis/api v0.0.0-20251029180050-ab9386a59fda/go.mod h1:fDMmzKV90WSg1NbozdqrE64fkuTv6mlq2zxo9ad+3yo=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260112192933-99fd39fd28a9 h1:IY6/YYRrFUk0JPp0xOVctvFIVuRnjccihY5kxf5g0TE=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260112192933-99fd39fd28a9/go.mod h1:j9x/tPzZkyxcgEFkiKEEGxfvyumM01BEtsW8xzOahRQ=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260114163908-3f89685c29c3 h1:C4WAdL+FbjnGlpp2S+HMVhBeCq2Lcib4xZqfPNF6OoQ=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260114163908-3f89685c29c3/go.mod h1:j9x/tPzZkyxcgEFkiKEEGxfvyumM01BEtsW8xzOahRQ=
google.golang.org/grpc v1.78.0 h1:K1XZG/yGDJnzMdd/uZHAkVqJE+xIDOcmdSFZkBUicNc=
google.golang.org/grpc v1.78.0/go.mod h1:I47qjTo4OKbMkjA/aOOwxDIiPSBofUtQUI5EfpWvW7U=
google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=

2
api/pkg/messaging/internal/natsb/NATS.go
View File

@@ -33,7 +33,7 @@ func (b *NatsBroker) Subscribe(event model.NotificationEvent) (<-chan me.Envelop
b.logger.Info("Subscribing to subject", zap.String("subject", subject))
// Create a bidirectional channel to send messages to
messageChan := make(chan me.Envelope)
messageChan := make(chan me.Envelope, b.bufferSize)
b.mu.Lock()
defer b.mu.Unlock()

22
api/pkg/messaging/internal/natsb/broker.go
View File

@@ -20,11 +20,12 @@ import (
type natsSubscriotions = map[string]*TopicSubscription
type NatsBroker struct {
nc *nats.Conn
js nats.JetStreamContext
logger *zap.Logger
topicSubs natsSubscriotions
mu sync.Mutex
nc *nats.Conn
js nats.JetStreamContext
logger *zap.Logger
topicSubs natsSubscriotions
mu sync.Mutex
bufferSize int
}
type envConfig struct {
@@ -32,6 +33,8 @@ type envConfig struct {
Port int
}
const defaultConsumerBufferSize = 1024
// loadEnv gathers and validates connection details from environment variables
// listed in the Settings struct. Invalid or missing values surface as a typed
// InvalidArgument error so callers can decide how to handle them.
@@ -83,6 +86,10 @@ func NewNatsBroker(logger mlogger.Logger, settings *nc.Settings) (*NatsBroker, e
var err error
var cfg *envConfig
var natsURL string
bufferSize := defaultConsumerBufferSize
if settings != nil && settings.BufferSize > 0 {
bufferSize = settings.BufferSize
}
if settings != nil && strings.TrimSpace(settings.URLEnv) != "" {
urlVal := strings.TrimSpace(os.Getenv(settings.URLEnv))
if urlVal != "" {
@@ -123,8 +130,9 @@ func NewNatsBroker(logger mlogger.Logger, settings *nc.Settings) (*NatsBroker, e
}
res := &NatsBroker{
logger: l.Named("nats"),
topicSubs: natsSubscriotions{},
logger: l.Named("nats"),
topicSubs: natsSubscriotions{},
bufferSize: bufferSize,
}
if res.nc, err = nats.Connect(natsURL, opts...); err != nil {

1
api/pkg/messaging/internal/natsb/config/config.go
View File

@@ -9,4 +9,5 @@ type Settings struct {
NATSName string `mapstructure:"broker_name" yaml:"broker_name"`
MaxReconnects int `mapstructure:"max_reconnects" yaml:"max_reconnects"`
ReconnectWait int `mapstructure:"reconnect_wait" yaml:"reconnect_wait"`
BufferSize int `mapstructure:"buffer_size" yaml:"buffer_size"`
}