Calling the right function upon invocation

girishsaraf · girishsaraf · commit 8b4b6ca9b642 · 2024-03-19T17:58:38.000-07:00
diff --git a/internal/stream/stream.go b/internal/stream/stream.go
@@ -2,37 +2,37 @@ package stream
 
 import (
 	"github.com/confluentinc/confluent-kafka-go/kafka"
+	"gostreambridge/internal/database/dsprocessors"
 	"gostreambridge/internal/queue/processors"
 	"log"
 	"os"
 	"os/signal"
 	"syscall"
-	"time"
 )
 
 func StartStreamBridge(bridgeDetails map[string]string) {
-	ConsumeMessages(bridgeDetails["upstreamApp"])
+	ConsumeMessages(bridgeDetails)
 }
 
-func ConsumeMessages(upstreamQueueType string) {
+func ConsumeMessages(bridgeDetails map[string]string) {
 
-	switch upstreamQueueType {
+	switch bridgeDetails["upstreamApp"] {
 	case "kafka":
 		// Create a Kafka message consumer
 		messages := processors.ConsumeKafkaMessages()
 		// Process each message
 		for msg := range messages {
-			ProcessKafkaMessage(msg)
+			ProcessKafkaMessage(bridgeDetails["downstreamApp"], msg)
 		}
 	case "rabbitmq":
 		// Create a Kafka message consumer
 		messages := processors.ConsumeAMQPMessages()
 		// Process each message
 		for msg := range messages {
-			ProcessAMQPMessage(msg)
+			ProcessAMQPMessage(bridgeDetails["downstreamApp"], msg)
 		}
 	default:
-		log.Fatalf("Unsupported message queue type: %s", upstreamQueueType)
+		log.Fatalf("Unsupported message queue type: %s", bridgeDetails["upstreamApp"])
 	}
 
 	// Wait for termination signal
@@ -42,15 +42,54 @@ func ConsumeMessages(upstreamQueueType string) {
 }
 
 // ProcessAMQPMessage Simulating message processing
-func ProcessAMQPMessage(msg []byte) {
-	// Simulating processing time
-	time.Sleep(2 * time.Second)
+func ProcessAMQPMessage(downstreamAppName string, msg []byte) {
+	// Calling the appropriate downstream flow
+	message := string(msg)
+	err := DetermineDownstreamFlow(downstreamAppName, message)
+	if err != nil {
+		return
+	}
 	log.Printf("Processed message: %s\n", string(msg))
 }
 
 // ProcessKafkaMessage Simulating message processing
-func ProcessKafkaMessage(msg *kafka.Message) {
-	// Simulating processing time
-	time.Sleep(2 * time.Second)
+func ProcessKafkaMessage(downstreamAppName string, msg *kafka.Message) {
+	// Calling the appropriate downstream flow
+	err := DetermineDownstreamFlow(downstreamAppName, string(msg.Value))
+	if err != nil {
+		return
+	}
 	log.Printf("Processed message: %s\n", string(msg.Value))
 }
+
+// DetermineDownstreamFlow to determine which function to call downstream
+func DetermineDownstreamFlow(downstreamAppName string, message string) error {
+	switch downstreamAppName {
+	case "mysql":
+		err := dsprocessors.WriteToMySQL(message)
+		if err != nil {
+			return err
+		}
+		return nil
+	case "sqlserver":
+		err := dsprocessors.WriteToSQLServer(message)
+		if err != nil {
+			return err
+		}
+		return nil
+	case "elastic":
+		err := dsprocessors.WriteToElastic(message)
+		if err != nil {
+			return err
+		}
+		return nil
+	case "kafka":
+		err := processors.WriteToKafka(message)
+		if err != nil {
+			return err
+		}
+		return nil
+	default:
+		return nil
+	}
+}
