Progress to get working Cassandra session

Author: Leo7Deng

backend/cassandra/service.go

@@ -0,0 +1,22 @@
+package cassandra
+
+import (
+	"fmt"
+	"github.com/Leo7Deng/ChatApp/models"
+	"github.com/gocql/gocql"
+	"github.com/google/uuid"
+)
+
+func InsertMessage(session *gocql.Session, message models.Message) error {
+	messageID := uuid.New().String()
+	query := session.Query(`
+		INSERT INTO chat_app.messages (circle_id, created_at, message_id, author_id, content)
+		VALUES (?, ?, ?, ?, ?)`,
+		message.CircleID, message.CreatedAt, messageID, message.AuthorID, message.Content)
+	err := query.Exec()
+	if err != nil {
+		fmt.Println("Error inserting message into Cassandra:", err)
+		return err
+	}
+	return nil
+}

backend/docker-compose.yml

@@ -65,13 +65,24 @@ services:
     networks:
       - backend-net
     environment:
+      CASSANDRA_START_RPC: "true"
+      CASSANDRA_RPC_ADDRESS: "0.0.0.0"
+      CASSANDRA_LISTEN_ADDRESS: "cassandra-node1"
+      CASSANDRA_ENDPOINT_SNITCH: "GossipingPropertyFileSnitch"
       CASSANDRA_CLUSTER_NAME: "cassandra-cluster"
-      CASSANDRA_SEEDS: "cassandra-node1,cassandra-node2,cassandra-node3"
-      CASSANDRA_NUM_TOKENS: 128
+      # CASSANDRA_SEEDS: "cassandra-node1,cassandra-node2"
+      CASSANDRA_NUM_TOKENS: 8
+      CASSANDRA_DC: "dc1"
     ports:
       - "9042:9042"
     volumes:
       - cassandra-data-node1:/var/lib/cassandra
+    healthcheck:
+      test: ["CMD-SHELL", "nodetool status"]
+      interval: 2m
+      timeout: 10s
+      retries: 3
+      start_period: 2m
 
   cassandra-node2:
     image: cassandra:latest
@@ -80,13 +91,25 @@ services:
     networks:
       - backend-net
     depends_on:
-      - cassandra-node1
+      cassandra-node1:
+        condition: service_healthy
     environment:
+      CASSANDRA_START_RPC: "true"
+      CASSANDRA_RPC_ADDRESS: "0.0.0.0"
+      CASSANDRA_LISTEN_ADDRESS: "cassandra-node2"
+      CASSANDRA_ENDPOINT_SNITCH: "GossipingPropertyFileSnitch"
       CASSANDRA_CLUSTER_NAME: "cassandra-cluster"
       CASSANDRA_SEEDS: "cassandra-node1"
-      CASSANDRA_NUM_TOKENS: 128
+      CASSANDRA_NUM_TOKENS: 8
+      CASSANDRA_DC: "dc1"
     volumes:
       - cassandra-data-node2:/var/lib/cassandra
+    healthcheck:
+      test: ["CMD-SHELL", "nodetool status"]
+      interval: 2m
+      timeout: 10s
+      retries: 3
+      start_period: 2m
 
   cassandra-node3:
     image: cassandra:latest
@@ -95,13 +118,25 @@ services:
     networks:
       - backend-net
     depends_on:
-      - cassandra-node1
+      cassandra-node2:
+        condition: service_healthy
     environment:
+      CASSANDRA_START_RPC: "true"
+      CASSANDRA_RPC_ADDRESS: "0.0.0.0"
+      CASSANDRA_LISTEN_ADDRESS: "cassandra-node3"
+      CASSANDRA_ENDPOINT_SNITCH: "GossipingPropertyFileSnitch"
       CASSANDRA_CLUSTER_NAME: "cassandra-cluster"
       CASSANDRA_SEEDS: "cassandra-node1"
-      CASSANDRA_NUM_TOKENS: 128
+      CASSANDRA_NUM_TOKENS: 8
+      CASSANDRA_DC: "dc1"
     volumes:
       - cassandra-data-node3:/var/lib/cassandra
+    healthcheck:
+      test: ["CMD-SHELL", "nodetool status"]
+      interval: 2m
+      timeout: 10s
+      retries: 3
+      start_period: 2m
   chat-app:
     build: .
     container_name: chat-app

backend/kafka/consumer.go

@@ -6,159 +6,95 @@ import (
 	"errors"
 	"fmt"
 	"log"
-	"os"
-	"os/signal"
 	"sync"
-	"syscall"
 
 	"github.com/IBM/sarama"
+	"github.com/Leo7Deng/ChatApp/cassandra"
 	"github.com/Leo7Deng/ChatApp/models"
 	"github.com/Leo7Deng/ChatApp/websockets"
-	// "github.com/gocql/gocql"
+	"github.com/gocql/gocql"
 )
 
 var hub *websockets.Hub
+var cassandraSession *gocql.Session
 
-// StartConsumer continuously logs incoming messages
-func StartConsumer(ctx context.Context) {
-	config := sarama.NewConfig()
-	config.Consumer.Return.Errors = true
-
-	client, err := sarama.NewConsumer(brokers, config)
-	if err != nil {
-		log.Fatalf("Failed to start Kafka consumer: %v", err)
-	}
-	defer client.Close()
-
-	partitionConsumer, err := client.ConsumePartition(topic, 0, sarama.OffsetNewest)
-	if err != nil {
-		log.Fatalf("Failed to consume Kafka topic: %v", err)
-	}
-	defer partitionConsumer.Close()
-
-	fmt.Println("Kafka Consumer started...")
-
-	for {
-		select {
-		case msg := <-partitionConsumer.Messages():
-			fmt.Printf("Kafka message: %s\n", string(msg.Value))
-			var message models.Message
-			err := json.Unmarshal(msg.Value, &message)
-			if err != nil {
-				fmt.Printf("Failed to unmarshal message: %v\n", err)
-			}
+func WebsocketConsumer(ctx context.Context, websocketHub *websockets.Hub) {
+	hub = websocketHub
+	groupID := "websocket-group"
+	handler := &WebsocketConsumerHandler{hub: websocketHub}
+	runConsumer(ctx , groupID, handler)
+}
 
-		case <-ctx.Done():
-			fmt.Println("Consumer shutting down...")
-			return
-		}
-	}
+func CassandraConsumer(ctx context.Context, session *gocql.Session) {
+	cassandraSession = session
+	groupID := "cassandra-group"
+	handler := &CassandraConsumerHandler{}
+	runConsumer(ctx, groupID, handler)
 }
 
-func WebsocketConsumer(ctx context.Context, websocketHub *websockets.Hub) {
-	hub = websocketHub
-	keepRunning := true
+func runConsumer(ctx context.Context, groupID string, handler sarama.ConsumerGroupHandler) {
 	config := sarama.NewConfig()
 	config.Consumer.Return.Errors = true
 	config.Consumer.Group.Rebalance.GroupStrategies = []sarama.BalanceStrategy{sarama.NewBalanceStrategyRoundRobin()}
+	config.Consumer.Offsets.Initial = sarama.OffsetNewest
 
-
-	consumer := Consumer{
-		ready: make(chan bool),
-	}
-
-	ctx, cancel := context.WithCancel(context.Background())
-	client, err := sarama.NewConsumerGroup(brokers, "websocket-group", config)
+	client, err := sarama.NewConsumerGroup(brokers, groupID, config)
 	if err != nil {
-		log.Panicf("Error creating consumer group client: %v", err)
+		log.Panicf("Error creating Kafka consumer group (%s): %v", groupID, err)
 	}
+	defer client.Close()
 
-	consumptionIsPaused := false
 	wg := &sync.WaitGroup{}
 	wg.Add(1)
+
 	go func() {
 		defer wg.Done()
-		for { 
-			err := client.Consume(ctx, []string{topic}, &consumer)
+		for {
+			err := client.Consume(ctx, []string{topic}, handler)
 			if err != nil {
 				if errors.Is(err, sarama.ErrClosedConsumerGroup) {
 					return
 				}
-				log.Printf("Error from consumer: %v", err)
+				log.Printf("Error from Kafka consumer (%s): %v", groupID, err)
 			}
 
 			// If context is canceled, stop the consumer
 			if ctx.Err() != nil {
 				return
 			}
-
-			// Reset readiness so we wait for the next session
-			consumer.ready = make(chan bool)
 		}
 	}()
 
-
-	<-consumer.ready // Await till the consumer has been set up
-	log.Println("Sarama consumer up and running!...")
-
-	sigusr1 := make(chan os.Signal, 1)
-	signal.Notify(sigusr1, syscall.SIGUSR1)
-
-	sigterm := make(chan os.Signal, 1)
-	signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)
-
-	for keepRunning {
-		select {
-		case <-ctx.Done():
-			log.Println("terminating: context cancelled")
-			keepRunning = false
-		case <-sigterm:
-			log.Println("terminating: via signal")
-			keepRunning = false
-		case <-sigusr1:
-			toggleConsumptionFlow(client, &consumptionIsPaused)
-		}
-	}
-	cancel()
+	log.Printf("%s consumer started...", groupID)
+	<-ctx.Done()
+	log.Printf("%s consumer shutting down...", groupID)
 	wg.Wait()
-	if err = client.Close(); err != nil {
-		log.Panicf("Error closing client: %v", err)
-	}
 }
 
-func toggleConsumptionFlow(client sarama.ConsumerGroup, isPaused *bool) {
-	if *isPaused {
-		client.ResumeAll()
-		log.Println("Resuming consumption")
-	} else {
-		client.PauseAll()
-		log.Println("Pausing consumption")
-	}
 
-	*isPaused = !*isPaused
-}
-
-// Consumer represents a Sarama consumer group consumer
-type Consumer struct {
-	ready chan bool
-}
+type WebsocketConsumerHandler struct {hub *websockets.Hub}
+type CassandraConsumerHandler struct {}
 
 // Setup is run at the beginning of a new session, before ConsumeClaim
-func (consumer *Consumer) Setup(sarama.ConsumerGroupSession) error {
-	// Mark the consumer as ready
-	close(consumer.ready)
+func (consumer *WebsocketConsumerHandler) Setup(sarama.ConsumerGroupSession) error {
+	return nil
+}
+func (consumer *CassandraConsumerHandler) Setup(sarama.ConsumerGroupSession) error {
 	return nil
 }
 
 // Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exited
-func (consumer *Consumer) Cleanup(sarama.ConsumerGroupSession) error {
+func (consumer *WebsocketConsumerHandler) Cleanup(sarama.ConsumerGroupSession) error {
+	return nil
+}
+func (consumer *CassandraConsumerHandler) Cleanup(sarama.ConsumerGroupSession) error {
 	return nil
 }
 
 // ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().
 // Once the Messages() channel is closed, the Handler must finish its processing
 // loop and exit.
-func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+func (consumer *WebsocketConsumerHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
 	// NOTE:
 	// Do not move the code below to a goroutine.
 	// The `ConsumeClaim` itself is called within a goroutine, see:
@@ -171,7 +107,7 @@ func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, clai
 				return nil
 			}
 			partition, offset := message.Partition, message.Offset
-			log.Printf("Kafka websocket consumer: %s | Partition: %d | Offset: %d\n", message.Value, partition, offset)
+			log.Printf("Kafka consumer: %s | Partition: %d | Offset: %d\n", message.Value, partition, offset)
 			var websocketMessage models.WebsocketMessage
 			err := json.Unmarshal(message.Value, &websocketMessage)
 			if err != nil {
@@ -189,46 +125,38 @@ func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, clai
 	}
 }
 
+func (consumer *CassandraConsumerHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+	for {
+		select {
+		case message, ok := <-claim.Messages():
+			if !ok {
+				log.Printf("message channel was closed")
+				return nil
+			}
+			partition, offset := message.Partition, message.Offset
+			var websocketMessage models.WebsocketMessage
+			err := json.Unmarshal(message.Value, &websocketMessage)
+			if err != nil {
+				fmt.Printf("Failed to unmarshal message: %v\n", err)
+			}
+			log.Printf("Cassandra consumer: %s | Partition: %d | Offset: %d\n", message.Value, partition, offset)
+			insertMessage := models.Message{
+				CircleID: websocketMessage.Message.CircleID,
+				AuthorID: websocketMessage.Message.AuthorID,
+				Content:  websocketMessage.Message.Content,
+				CreatedAt: websocketMessage.Message.CreatedAt,
+			}
+			err = cassandra.InsertMessage(cassandraSession, insertMessage)
 
-
-
-
-
-
-
-
-
-
-
-
-
-// 	//////////////////
-// 	group, err := sarama.NewConsumerGroup(brokers, "websocket-group", config)
-// 	if err != nil {
-// 		log.Fatalf("Failed to start Kafka consumer group: %v", err)
-// 	}
-// 	defer group.Close()
-
-// 	fmt.Println("Kafka Consumer started...")
-
-// 	for {
-// 		select {
-// 		case msg := <-partitionConsumer.Messages():
-// 			var websocketMessage models.WebsocketMessage
-// 			fmt.Printf("Kafka consumer viewed: %s\n", string(msg.Value))
-// 			err := json.Unmarshal(msg.Value, &websocketMessage)
-// 			if err != nil {
-// 				fmt.Printf("Failed to unmarshal message: %v\n", err)
-// 			}
-// 			websocketMessage.Origin = "server"
-// 			hub.SendWebsocketMessage(websocketMessage)
-// 		case <-ctx.Done():
-// 			fmt.Println("Consumer shutting down...")
-// 			return
-// 		}
-// 	}
-// }
-
-// func CassandraConsumer(ctx context.Context, cassandraSession *gocql.Session) {
-
-// }
+			// Handle error on unprocessed message insert into Cassandra
+			if err != nil {
+				fmt.Printf("Failed to insert message: %v\n", err)
+			} else {
+				log.Printf("Message inserted into Cassandra: %v\n", insertMessage)
+				session.MarkMessage(message, "")
+			}
+		case <-session.Context().Done():
+			return nil
+		}
+	}
+}