Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Connection recovery and multiple URIs #27

Merged
merged 4 commits into from
Aug 13, 2024
Merged

Connection recovery and multiple URIs #27

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8358484
AMQP 1.0 connection recovery
mkuratczyk Aug 12, 2024
8bda80c
Allow multiple URIs
mkuratczyk Aug 12, 2024
9ac0b3e
Allow URIs without scheme
mkuratczyk Aug 13, 2024
948d03d
STOMP connection recovery
mkuratczyk Aug 13, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 6 additions & 6 deletions cmd/root.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -186,8 +186,8 @@ func RootCmd() *cobra.Command {
}
rootCmd.PersistentFlags().
VarP(enumflag.New(&log.Level, "log-level", log.Levels, enumflag.EnumCaseInsensitive), "log-level", "l", "Log level (debug, info, error)")
rootCmd.PersistentFlags().StringVarP(&cfg.PublisherUri, "publisher-uri", "", "", "URI for publishing")
rootCmd.PersistentFlags().StringVarP(&cfg.ConsumerUri, "consumer-uri", "", "", "URI for consuming")
rootCmd.PersistentFlags().StringSliceVarP(&cfg.PublisherUri, "publisher-uri", "", nil, "URI for publishing")
rootCmd.PersistentFlags().StringSliceVarP(&cfg.ConsumerUri, "consumer-uri", "", nil, "URI for consuming")
rootCmd.PersistentFlags().IntVarP(&cfg.Publishers, "publishers", "x", 1, "The number of publishers to start")
rootCmd.PersistentFlags().IntVarP(&cfg.Consumers, "consumers", "y", 1, "The number of consumers to start")
rootCmd.PersistentFlags().
Expand Down Expand Up @@ -309,13 +309,13 @@ func setUris(cfg *config.Config, command string) {
if command == "version" {
return
}
if cfg.PublisherUri == "" {
if cfg.PublisherUri == nil {
println("setting publisher uri to ", defaultUri(strings.Split(command, "-")[0]))
(*cfg).PublisherUri = defaultUri(strings.Split(command, "-")[0])
(*cfg).PublisherUri = []string{defaultUri(strings.Split(command, "-")[0])}
}
if cfg.ConsumerUri == "" {
if cfg.ConsumerUri == nil {
println("setting consumer uri to ", defaultUri(strings.Split(command, "-")[1]))
(*cfg).ConsumerUri = defaultUri(strings.Split(command, "-")[1])
(*cfg).ConsumerUri = []string{defaultUri(strings.Split(command, "-")[1])}
}
}

Expand Down
122 changes: 86 additions & 36 deletions pkg/amqp10_client/consumer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,42 +24,77 @@ type Amqp10Consumer struct {
Id int
Connection *amqp.Conn
Session *amqp.Session
Receiver *amqp.Receiver
Topic string
Config config.Config
whichUri int
}

func NewConsumer(cfg config.Config, id int) *Amqp10Consumer {
// open connection
hostname, vhost := hostAndVHost(cfg.ConsumerUri)
conn, err := amqp.Dial(context.TODO(), cfg.ConsumerUri, &amqp.ConnOptions{
HostName: vhost,
TLSConfig: &tls.Config{
ServerName: hostname}})
if err != nil {
log.Error("consumer failed to connect", "protocol", "amqp-1.0", "consumerId", id, "error", err.Error())
return nil
consumer := &Amqp10Consumer{
Id: id,
Connection: nil,
Session: nil,
Receiver: nil,
Topic: topic.CalculateTopic(cfg.ConsumeFrom, id),
Config: cfg,
whichUri: 0,
}

// open seesion
session, err := conn.NewSession(context.TODO(), nil)
if err != nil {
log.Error("consumer failed to create a session", "protocol", "amqp-1.0", "consumerId", id, "error", err.Error())
return nil
}
// TODO
consumer.Connect(context.TODO())

topic := topic.CalculateTopic(cfg.ConsumeFrom, id)
return consumer
}

return &Amqp10Consumer{
Id: id,
Connection: conn,
Session: session,
Topic: topic,
Config: cfg,
func (c *Amqp10Consumer) Connect(ctx context.Context) {
if c.Receiver != nil {
_ = c.Receiver.Close(ctx)
}
if c.Session != nil {
_ = c.Session.Close(context.Background())
}
if c.Connection != nil {
_ = c.Connection.Close()
}
c.Receiver = nil
c.Session = nil
c.Connection = nil

for c.Connection == nil {
if c.whichUri >= len(c.Config.ConsumerUri) {
c.whichUri = 0
}
uri := c.Config.ConsumerUri[c.whichUri]
c.whichUri++
hostname, vhost := hostAndVHost(uri)
conn, err := amqp.Dial(context.TODO(), uri, &amqp.ConnOptions{
HostName: vhost,
TLSConfig: &tls.Config{
ServerName: hostname,
},
})
if err != nil {
log.Error("consumer failed to connect", "protocol", "amqp-1.0", "consumerId", c.Id, "error", err.Error())
time.Sleep(1 * time.Second)
} else {
log.Debug("consumer connected", "protocol", "amqp-1.0", "consumerId", c.Id, "uri", uri)
c.Connection = conn
}
}

for c.Session == nil {
session, err := c.Connection.NewSession(context.TODO(), nil)
if err != nil {
log.Error("consumer failed to create a session", "protocol", "amqp-1.0", "consumerId", c.Id, "error", err.Error())
time.Sleep(1 * time.Second)
} else {
c.Session = session
}
}
}

func (c Amqp10Consumer) Start(ctx context.Context, subscribed chan bool) {
func (c *Amqp10Consumer) CreateReceiver(ctx context.Context) {
var durability amqp.Durability
switch c.Config.QueueDurability {
case config.None:
Expand All @@ -69,29 +104,42 @@ func (c Amqp10Consumer) Start(ctx context.Context, subscribed chan bool) {
case config.UnsettledState:
durability = amqp.DurabilityUnsettledState
}
receiver, err := c.Session.NewReceiver(ctx, c.Topic, &amqp.ReceiverOptions{SourceDurability: durability, Credit: int32(c.Config.ConsumerCredits), Properties: buildLinkProperties(c.Config), Filters: buildLinkFilters(c.Config)})
if err != nil {
log.Error("consumer failed to create a receiver", "protocol", "amqp-1.0", "consumerId", c.Id, "error", err.Error())
return

for c.Receiver == nil {
receiver, err := c.Session.NewReceiver(ctx, c.Topic, &amqp.ReceiverOptions{SourceDurability: durability, Credit: int32(c.Config.ConsumerCredits), Properties: buildLinkProperties(c.Config), Filters: buildLinkFilters(c.Config)})
if err != nil {
log.Error("consumer failed to create a receiver", "protocol", "amqp-1.0", "consumerId", c.Id, "error", err.Error())
time.Sleep(1 * time.Second)
} else {
c.Receiver = receiver
}
}
close(subscribed)
log.Debug("consumer subscribed", "protocol", "amqp-1.0", "consumerId", c.Id, "terminus", c.Topic, "durability", durability)
}

func (c *Amqp10Consumer) Start(ctx context.Context, subscribed chan bool) {
m := metrics.EndToEndLatency

c.CreateReceiver(ctx)
close(subscribed)
log.Info("consumer started", "protocol", "amqp-1.0", "consumerId", c.Id, "terminus", c.Topic)
previousMessageTimeSent := time.Unix(0, 0)

for i := 1; i <= c.Config.ConsumeCount; i++ {
for i := 1; i <= c.Config.ConsumeCount; {
if c.Receiver == nil {
c.CreateReceiver(ctx)
log.Debug("consumer subscribed", "protocol", "amqp-1.0", "consumerId", c.Id, "terminus", c.Topic)
}

select {
case <-ctx.Done():
c.Stop("time limit reached")
return
default:
msg, err := receiver.Receive(ctx, nil)
msg, err := c.Receiver.Receive(ctx, nil)
if err != nil {
log.Error("failed to receive a message", "protocol", "amqp-1.0", "consumerId", c.Id, "terminus", c.Topic)
return
c.Connect(ctx)
continue
}

payload := msg.GetData()
Expand All @@ -112,20 +160,22 @@ func (c Amqp10Consumer) Start(ctx context.Context, subscribed chan bool) {
time.Sleep(c.Config.ConsumerLatency)
}

err = receiver.AcceptMessage(ctx, msg)
err = c.Receiver.AcceptMessage(ctx, msg)
if err != nil {
log.Error("message NOT accepted", "protocol", "amqp-1.0", "consumerId", c.Id, "terminus", c.Topic)
} else {
metrics.MessagesConsumed.With(prometheus.Labels{"protocol": "amqp-1.0", "priority": priority}).Inc()
i++
log.Debug("message accepted", "protocol", "amqp-1.0", "consumerId", c.Id, "terminus", c.Topic)
}
metrics.MessagesConsumed.With(prometheus.Labels{"protocol": "amqp-1.0", "priority": priority}).Inc()
log.Debug("message accepted", "protocol", "amqp-1.0", "consumerId", c.Id, "terminus", c.Topic)
}
}

c.Stop("message count reached")
log.Debug("consumer finished", "protocol", "amqp-1.0", "consumerId", c.Id)
}

func (c Amqp10Consumer) Stop(reason string) {
func (c *Amqp10Consumer) Stop(reason string) {
log.Debug("closing connection", "protocol", "amqp-1.0", "consumerId", c.Id, "reason", reason)
_ = c.Connection.Close()
}
Expand Down
Loading
Loading