.

gitea-events-rabbitmq-publisher/rabbitmq.go

@@ -6,7 +6,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"io"
 	"log"
 	"time"
 
@@ -26,32 +25,44 @@ type Message struct {
 
 var messageQueue chan Message
 
-func PublishMessage(giteaWebhookType, giteaOrg string, msgBody io.Reader) error {
+func PublicActionMessage(giteaOrg, giteaWebhookType, action string, data []byte) error {
 	if messageQueue == nil {
 		return fmt.Errorf("Queue not initialized")
 	}
 
-	data, err := io.ReadAll(msgBody)
-	if err != nil {
-		return fmt.Errorf("Error reading JSON data. Err: %v", err)
-	}
-
 	if !json.Valid(data) {
 		return errors.New("Invalid JSON in request")
 	}
 
-	select {
-	case messageQueue <- Message{
-		Topic: fmt.Sprintf("opensuse.gitea.%s.%s", giteaOrg, giteaWebhookType),
-		Body:  data,
-	}:
+	var msg Message
+	if len(action) > 0 {
+		msg = Message{
+			Topic: fmt.Sprintf("opensuse.gitea.%s.%s.%s", giteaOrg, giteaWebhookType, action),
+			Body:  data,
+		}
+	} else {
+		msg = Message{
+			Topic: fmt.Sprintf("opensuse.gitea.%s.%s", giteaOrg, giteaWebhookType),
+			Body:  data,
+		}
+	}
 
+	if len(action) > 0 {
+		msg.Topic = msg.Topic + "." + action
+	}
+
+	select {
+	case messageQueue <- msg:
 	default:
 		return errors.New("Enable to queue message. Possibly queue full.")
 	}
 	return nil
 }
 
+func PublishMessage(giteaWebhookType, giteaOrg string, data []byte) error {
+	return PublicActionMessage(giteaOrg, giteaWebhookType, "", data)
+}
+
 func ConnectToExchangeForPublish(host, username, password string) {
 	defer func() {
 		if r := recover(); r != nil {