.

gitea-events-rabbitmq-publisher/main.go

@@ -1,21 +1,37 @@
 package main
 
 import (
+	"fmt"
 	"log"
 	"net/http"
+	"os"
 
 	"src.opensuse.org/autogits/common"
 )
 
 const (
-	ListenAddr          = "[::1]:8001"
+	ListenAddr          = "[::1]:8002"
 	RabbitForwarderPath = "rabbitmq-forwarder"
 )
 
+func connectToRabbitMQ() {
+	host := os.Getenv("RABBITMQ_HOST")
+	username := os.Getenv("RABBITMQ_USERNAME")
+	password := os.Getenv("RABBITMQ_PASSWORD")
+
+	if len(host) == 0 || len(username) == 0 || len(password) == 0 {
+		fmt.Printf("Missing RABBITMQ_HOST, RABBITMQ_USERNAME, RABBITMQ_PASSWORD")
+		os.Exit(1)
+	}
+	
+	ConnectToExchangeForPublish(host, username, password)
+}
+
 func main() {
 	common.RequireGiteaSecretToken()
+	connectToRabbitMQ()
 
-	http.HandleFunc("/"+RabbitForwarderPath, func(res http.ResponseWriter, req *http.Request) {
+	http.HandleFunc("POST /rabbitmq-forwarder/{Org}", func(res http.ResponseWriter, req *http.Request) {
 		if len(req.Header.Get("Content-Type")) == 0 ||
 			req.Header["Content-Type"][0] != "application/json" ||
 			req.Method != "POST" {
@@ -32,7 +48,7 @@ func main() {
 			return
 		}
 		reqType := hdr[0]
-		err := PublicMessage(reqType, req.Body)
+		err := PublicMessage(reqType, req.PathValue("Org"), req.Body)
 		if err != nil {
 			log.Printf("hook (%s) processing error: %v\n", reqType, err)
 			res.WriteHeader(http.StatusBadRequest)

gitea-events-rabbitmq-publisher/rabbitmq.go

@@ -20,20 +20,30 @@ func failOnError(err error, msg string) {
 
 type Message struct {
 	Topic string
-	Body string
+	Body  []byte
 }
 
-var messageQueue chan Messag
+var messageQueue chan Message
+
+func PublicMessage(giteaWebhookType, giteaOrg string, msgBody io.Reader) error {
+	if messageQueue == nil {
+		return fmt.Errorf("Queue not initialized")
+	}
 
-func PublicMessage(giteaWebhookType string, msgBody io.Reader) error {
 	data, err := io.ReadAll(msgBody)
+	if err != nil {
+		return fmt.Errorf("Error reading JSON data. Err: %v", err)
+	}
+
 	if !json.Valid(data) {
 		return fmt.Errorf("Invalid JSON in request")
 	}
 
 	messageQueue <- Message{
-		Topic: "opensuse.gitea." + giteaWebhookType + "." + 
+		Topic: fmt.Sprintf("opensuse.gitea.%s.%s", giteaOrg, giteaWebhookType),
+		Body:  data,
 	}
+	return nil
 }
 
 func ConnectToExchangeForPublish(host, username, password string) {
@@ -44,6 +54,7 @@ func ConnectToExchangeForPublish(host, username, password string) {
 			go ConnectToExchangeForPublish(host, username, password)
 		}
 	}()
+
 	if messageQueue == nil {
 		messageQueue = make(chan Message, 10240)
 	}
@@ -69,22 +80,32 @@ func ConnectToExchangeForPublish(host, username, password string) {
 	for {
 		msg, ok := <-messageQueue
 		if !ok {
-			log.Printf("channel/connection closed?\n")
+			log.Printf("Shutdown ... \n")
-
-			if connection.IsClosed() {
-				// reconnect
-				log.Printf("reconnecting...")
-				time.Sleep(5 * time.Second)
-				go ConnectToExchangeForPublish(host, username, password)
-			}
 			return
 		}
 
-		ch.Publish("pubsub", msg.Topic, false, false, rabbitmq.Publishing {
+		err = ch.Publish("pubsub", msg.Topic, false, false, rabbitmq.Publishing{
 			ContentType: "application/json",
 			AppId:       RabbitForwarderPath,
 			Body:        []byte(msg.Body),
 		})
-	}
-}
 
+		if err != nil {
+			if connection.IsClosed() {
+				select {
+				case messageQueue <- msg:
+					log.Printf("requed ...")
+				default:
+					log.Printf("queue full... message lost")
+				}
+				// reconnect
+				log.Printf("reconnecting...")
+				time.Sleep(5 * time.Second)
+				go ConnectToExchangeForPublish(host, username, password)
+				return
+			} else {
+				log.Printf("Error sending request. %v", err)
+			}
+		}
+	}
+}