Configurable AMQP1.0 consumer credits

rabbitmq · Sep 26, 2023 · 5f3ca27 · 5f3ca27
1 parent daebb48
commit 5f3ca27
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 1 deletion.
diff --git a/cmd/root.go b/cmd/root.go
@@ -50,6 +50,7 @@ func RootCmd() *cobra.Command {
 			start(cfg, common.AMQP, common.AMQP)
 		},
 	}
+	amqp_amqp.Flags().IntVarP(&cfg.Amqp.ConsumerCredits, "amqp-consumer-credits", "", 1, "AMQP 1.0 consumer credits")
 
 	amqp_stomp = &cobra.Command{
 		Use: "amqp-stomp",
@@ -79,6 +80,7 @@ func RootCmd() *cobra.Command {
 			start(cfg, common.STOMP, common.AMQP)
 		},
 	}
+	stomp_amqp.Flags().IntVarP(&cfg.Amqp.ConsumerCredits, "amqp-consumer-credits", "", 1, "AMQP 1.0 consumer credits")
 
 	stomp_mqtt = &cobra.Command{
 		Use: "stomp-mqtt",
@@ -101,6 +103,7 @@ func RootCmd() *cobra.Command {
 			start(cfg, common.MQTT, common.AMQP)
 		},
 	}
+	mqtt_amqp.Flags().IntVarP(&cfg.Amqp.ConsumerCredits, "amqp-consumer-credits", "", 1, "AMQP 1.0 consumer credits")
 
 	mqtt_stomp = &cobra.Command{
 		Use: "mqtt-stomp",

diff --git a/pkg/amqp10_client/consumer.go b/pkg/amqp10_client/consumer.go
@@ -49,7 +49,7 @@ func NewConsumer(cfg config.Config, id int) *Amqp10Consumer {
 }
 
 func (c Amqp10Consumer) Start(subscribed chan bool) {
-	receiver, err := c.Session.NewReceiver(context.TODO(), c.Topic, &amqp.ReceiverOptions{Durability: amqp.DurabilityUnsettledState})
+	receiver, err := c.Session.NewReceiver(context.TODO(), c.Topic, &amqp.ReceiverOptions{Durability: amqp.DurabilityUnsettledState, Credit: int32(c.Config.Amqp.ConsumerCredits)})
 	if err != nil {
 		log.Error("consumer failed to create a receiver", "protocol", "amqp-1.0", "consumerId", c.Id, "error", err.Error())
 		return

diff --git a/pkg/config/config.go b/pkg/config/config.go
@@ -2,6 +2,10 @@ package config
 
 import "time"
 
+type AmqpOptions struct {
+	ConsumerCredits int
+}
+
 type Config struct {
 	PublisherUri    string
 	ConsumerUri     string
@@ -15,4 +19,5 @@ type Config struct {
 	Rate            int
 	Duration        time.Duration
 	UseMillis       bool
+	Amqp            AmqpOptions
 }