Add Cassandra cluster and initialize session. Able to get cassandra nodes running in cluster

Author: Leo7Deng

backend/cassandra/client.go

@@ -8,20 +8,17 @@ import (
 	"github.com/gocql/gocql"
 )
 
-func CassandraClient() {
+var (
+	cluster *gocql.ClusterConfig
+	session *gocql.Session
+)
+
+func CassandraSession() *gocql.Session {
 	fmt.Println("Creating Cassandra session")
-	cluster := gocql.NewCluster("cassandra-node1:9042")
-	// cluster.Authenticator = gocql.PasswordAuthenticator{
-	// 	Username: "user",
-	// 	Password: "password"
-	// }
-	cluster.ProtoVersion = 4
-	// cluster.Keyspace = "keyspace"
-	// cluster.Port = 29042
-	// cluster.CQLVersion = "5.0.0"
-	// cluster.ConnectTimeout = time.Second * 6
+	cluster = gocql.NewCluster("cassandra-node1:9042", "cassandra-node2:9042", "cassandra-node3:9042")
+	var err error
+	cluster.Consistency = gocql.Quorum
 	cluster.PoolConfig.HostSelectionPolicy = gocql.TokenAwareHostPolicy(gocql.DCAwareRoundRobinPolicy("cassandra-node1"))
-	fmt.Println("3")
 	session, err := cluster.CreateSession()
 	if err != nil {
 		log.Fatal(err)
@@ -31,24 +28,23 @@ func CassandraClient() {
 	for _, host := range cluster.Hosts {
 		fmt.Println("- Node:", host)
 	}
-	// session, err := cluster.CreateSession()
-	// if err != nil {
-	// 	panic(err)
-	// }
-	// defer session.Close()
-	// fmt.Println("5")
-	// fmt.Println("Connected to Cassandra Nodes:")
-	// for _, host := range cluster.Hosts {
-	// 	fmt.Println("- Node:", host)
-	// }
 
 	// Print which node is being used for a query
-	// 	var clusterID string
-	// 	iter := session.Query("SELECT release_version FROM system.local").Iter()
-	// 	for iter.Scan(&clusterID) {
-	// 		fmt.Println("Running query on:", clusterID)
-	// 	}
-	// 	if err := iter.Close(); err != nil {
-	// 		log.Fatal("Error closing iterator:", err)
-	// 	}
+	var clusterID string
+	iter := session.Query("SELECT release_version FROM system.local").Iter()
+	for iter.Scan(&clusterID) {
+		fmt.Println("Running query on node ", clusterID)
+	}
+	if err := iter.Close(); err != nil {
+		log.Fatal("Error closing iterator:", err)
+	}
+
+	return session
 }
+
+func Close() {
+	if session != nil {
+		session.Close()
+		fmt.Println("Closing Cassandra session")
+	}
+}

backend/main.go

@@ -56,17 +56,18 @@ func main() {
 	postgres.ConnectPSQL()
 	defer postgres.ClosePSQL()
 
+	cassandraSession := cassandra.CassandraSession()
+	defer cassandraSession.Close()
+
 	hub := websockets.NewHub()
 
 	ctx, cancel := context.WithCancel(context.Background())
 	var wg sync.WaitGroup
-	kafka.InitKafka(ctx, &wg, hub)
+	kafka.InitKafka(ctx, &wg, hub, cassandraSession)
 
 	go hub.Run() 
 	fmt.Println("Websocket server started", hub)
 
-	cassandra.CassandraClient()
-
 	mux.HandleFunc("/api/register", middleware.AddCorsHeaders(auth.RegisterHandler))
 	mux.HandleFunc("/api/login", middleware.AddCorsHeaders(auth.LoginHandler))
 	mux.Handle("/api/circles", middleware.AddCorsHeaders(

frontend/app/context/authContext.tsx

@@ -19,7 +19,6 @@ export const AuthProvider: React.FC<{ children: React.ReactNode }> = ({ children
     const [shouldRedirect, setShouldRedirect] = useState(false);
 
     async function getAccessToken(): Promise<string> {
-        console.log(Date.now(), expiryTime);
         if (!accessToken || Date.now() >= expiryTime) {
             const token = await refreshAccessToken();
             return token;