fix(iris): add message type when publishing message

skkuding · Sep 7, 2024 · 6200d16 · 6200d16
1 parent f1d7031
commit 6200d16
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 4 deletions.
diff --git a/apps/iris/src/connector/rabbitmq/connector.go b/apps/iris/src/connector/rabbitmq/connector.go
@@ -83,7 +83,7 @@ func (c *connector) handle(message amqp.Delivery, ctx context.Context) {
 	}
 
 	for result := range resultChan {
-		if err := c.producer.Publish(result, ctx); err != nil {
+		if err := c.producer.Publish(result, ctx, message.Type); err != nil {
 			c.logger.Log(logger.ERROR, fmt.Sprintf("failed to publish result: %s: %s", string(result), err))
 			// nack
 		} else {

diff --git a/apps/iris/src/connector/rabbitmq/producer.go b/apps/iris/src/connector/rabbitmq/producer.go
@@ -10,7 +10,7 @@ import (
 
 type Producer interface {
 	OpenChannel() error
-	Publish([]byte, context.Context) error
+	Publish([]byte, context.Context, string) error
 	CleanUp() error
 }
 
@@ -99,7 +99,7 @@ func (p *producer) confirmHandler(confirms chan amqp.Confirmation) {
 	}
 }
 
-func (p *producer) Publish(result []byte, ctx context.Context) error {
+func (p *producer) Publish(result []byte, ctx context.Context, messageType string) error {
 
 	seqNo := p.channel.GetNextPublishSeqNo()
 	p.logger.Log(logger.INFO, fmt.Sprintf("publishing %dB body", len(result)))
@@ -117,7 +117,8 @@ func (p *producer) Publish(result []byte, ctx context.Context) error {
 			ContentEncoding: "",
 			Body:            result,
 			DeliveryMode:    amqp.Persistent, // 1=non-persistent, 2=persistent
-			Priority:        0,               // 0-9
+			Priority:        0,
+			Type:            messageType, // 0-9
 		},
 	); err != nil {
 		return fmt.Errorf("exchange publish: %s", err)