Add --amqp-subject option

rabbitmq · Oct 3, 2023 · 28f0d27 · 28f0d27
1 parent fb18dfe
commit 28f0d27
Show file tree

Hide file tree

Showing 3 changed files with 21 additions and 16 deletions.
diff --git a/cmd/root.go b/cmd/root.go
@@ -152,6 +152,7 @@ func RootCmd() *cobra.Command {
 	rootCmd.PersistentFlags().DurationVarP(&cfg.Duration, "duration", "z", 0, "Duration (eg. 10s, 5m, 2h)")
 	rootCmd.PersistentFlags().BoolVarP(&cfg.UseMillis, "use-millis", "m", false, "Use milliseconds for timestamps")
 	rootCmd.PersistentFlags().VarP(enumflag.New(&cfg.QueueDurability, "queue-durability", config.AmqpDurabilityModes, enumflag.EnumCaseInsensitive), "queue-durability", "", "Queue durability (default: configuration - the queue definition is durable)")
+	rootCmd.PersistentFlags().StringVar(&cfg.Amqp.Subject, "amqp-subject", "", "AMQP 1.0 message subject")
 	rootCmd.PersistentFlags().BoolVarP(&cfg.MessageDurability, "message-durability", "d", true, "Mark messages as durable (default=true)")
 
 	rootCmd.AddCommand(amqp_amqp)

diff --git a/pkg/amqp10_client/publisher.go b/pkg/amqp10_client/publisher.go
@@ -117,6 +117,9 @@ func (p Amqp10Publisher) StartRateLimited() {
 func (p Amqp10Publisher) Send() {
 	utils.UpdatePayload(p.Config.UseMillis, &p.msg)
 	msg := amqp.NewMessage(p.msg)
+	if p.Config.Amqp.Subject != "" {
+		msg.Properties = &amqp.MessageProperties{Subject: &p.Config.Amqp.Subject}
+	}
 	msg.Header = &amqp.MessageHeader{
 		Durable: p.Config.MessageDurability}
 	timer := prometheus.NewTimer(metrics.PublishingLatency.With(prometheus.Labels{"protocol": "amqp-1.0"}))

diff --git a/pkg/config/config.go b/pkg/config/config.go
@@ -22,6 +22,7 @@ var AmqpDurabilityModes = map[AmqpDurabilityMode][]string{
 
 type AmqpOptions struct {
 	ConsumerCredits int
+	Subject string
 }
 
 type MqttOptions struct {
@@ -30,23 +31,23 @@ type MqttOptions struct {
 }
 
 type Config struct {
-	PublisherUri    string
-	ConsumerUri     string
-	Publishers      int
-	Consumers       int
-	PublishCount    int
-	ConsumeCount    int
-	PublishTo       string
-	ConsumeFrom     string
-	Size            int
-	Rate            int
-	Duration        time.Duration
-	UseMillis       bool
-	QueueDurability AmqpDurabilityMode
+	PublisherUri      string
+	ConsumerUri       string
+	Publishers        int
+	Consumers         int
+	PublishCount      int
+	ConsumeCount      int
+	PublishTo         string
+	ConsumeFrom       string
+	Size              int
+	Rate              int
+	Duration          time.Duration
+	UseMillis         bool
+	QueueDurability   AmqpDurabilityMode
 	MessageDurability bool
-	Amqp            AmqpOptions
-	MqttPublisher   MqttOptions
-	MqttConsumer    MqttOptions
+	Amqp              AmqpOptions
+	MqttPublisher     MqttOptions
+	MqttConsumer      MqttOptions
 }
 
 func NewConfig() Config {