Merge pull request #50 from MerzMax/feature/45_chat_rooms

Feature/45 chat rooms

grafana/dashboards/cadvisor-exporter.json

@@ -23,7 +23,7 @@
   "fiscalYearStartMonth": 0,
   "gnetId": 14282,
   "graphTooltip": 0,
-  "iteration": 1642865978052,
+  "iteration": 1643972821121,
   "links": [],
   "liveNow": false,
   "panels": [
@@ -86,7 +86,7 @@
       "targets": [
         {
           "exemplar": true,
-          "expr": "sum(rate(container_cpu_usage_seconds_total{instance=~\"$host\",name=~\".+server.+\",name=~\".+\"}[5m])) *100",
+          "expr": "sum(rate(container_cpu_usage_seconds_total{instance=~\"$host\",name=~\".+server.+\",name=~\".+\"}[60s])) *100",
           "hide": false,
           "interval": "",
           "legendFormat": "Server",
@@ -431,7 +431,8 @@
       "steppedLine": false,
       "targets": [
         {
-          "expr": "sum(rate(container_network_receive_bytes_total{instance=~\"$host\",name=~\"$container\",name=~\".+\"}[5m])) by (name)",
+          "exemplar": true,
+          "expr": "sum(rate(container_network_receive_bytes_total{instance=~\"$host\",name=~\"$container\",name=~\".+\"}[60s])) by (name)",
           "hide": false,
           "interval": "",
           "legendFormat": "{{name}}",
@@ -525,7 +526,8 @@
       "steppedLine": false,
       "targets": [
         {
-          "expr": "sum(rate(container_network_transmit_bytes_total{instance=~\"$host\",name=~\"$container\",name=~\".+\"}[5m])) by (name)",
+          "exemplar": true,
+          "expr": "sum(rate(container_network_transmit_bytes_total{instance=~\"$host\",name=~\"$container\",name=~\".+\"}[60s])) by (name)",
           "interval": "",
           "legendFormat": "{{name}}",
           "refId": "A"

src/client/client.go

@@ -26,6 +26,7 @@ type Client struct {
 	MsgSize          int
 	MsgFrequency     int
 	MsgEvents        chan<- *MessageEventEntry
+	Room             string
 }
 
 func (client *Client) Start() error {
@@ -34,18 +35,28 @@ func (client *Client) Start() error {
 	if client.IsLoadTestClient {
 		client.id = uuid.New().String()
 	} else {
-		log.Printf("Client started in loadtest mode. Please input your id: ")
+		log.Printf("Client started in loadtest mode.")
+		log.Printf("Please input your id:")
 		input, err := consoleReader.ReadString('\n')
 		// convert CRLF to LF
 		client.id = strings.Replace(input, "\n", "", -1)
 		if err != nil || len(client.id) < 1 {
-			log.Printf("Failed to read the name input. Using default id: MuM")
+			log.Printf("Failed to read the name input. Using default name: MuM")
 			client.id = "MuM"
 		}
+
+		log.Printf("Please input your chat room:")
+		input, err = consoleReader.ReadString('\n')
+		// convert CRLF to LF
+		client.Room = strings.Replace(input, "\n", "", -1)
+		if err != nil || len(client.Room) < 1 {
+			log.Printf("Failed to read the chat room input. Using default id: test")
+			client.Room = "test"
+		}
 	}
 
 	// Connection Establishment
-	wsConnection, _, err := websocket.DefaultDialer.Dial(client.ServerUrl, nil)
+	wsConnection, _, err := websocket.DefaultDialer.Dial(client.ServerUrl+"/"+client.Room, nil)
 	if err != nil {
 		log.Fatal("Error connecting to Websocket Server:", err)
 	}
@@ -127,7 +138,7 @@ func (client *Client) receiveHandler(ctx context.Context, waitGroup *sync.WaitGr
 				continue
 			}
 
-			// If the loadtest mode is activated, there will be added a new message event with the metadata of this message.
+			// If the load test mode is activated, there will be added a new message event with the metadata of this message.
 			if client.IsLoadTestClient {
 				var msgEventEntry = MessageEventEntry{
 					ClientId:  client.id,
@@ -137,9 +148,9 @@ func (client *Client) receiveHandler(ctx context.Context, waitGroup *sync.WaitGr
 					Type:      Received,
 				}
 				client.MsgEvents <- &msgEventEntry
-
-				log.Printf("%v", message)
 			}
+
+			log.Printf("%v", message)
 		}
 	}
 }
@@ -193,7 +204,7 @@ func (client *Client) sendHandler(ctx context.Context, waitGroup *sync.WaitGroup
 				return
 			}
 
-			// If the loadtest mode is activated, there will be added a new message event with the metadata of this message.
+			// If the load test mode is activated, there will be added a new message event with the metadata of this message.
 			if client.IsLoadTestClient {
 				var msgEventEntry = MessageEventEntry{
 					ClientId:  client.id,

src/go.mod

@@ -5,6 +5,7 @@ go 1.17
 require (
 	github.com/go-echarts/go-echarts/v2 v2.2.4
 	github.com/google/uuid v1.3.0
+	github.com/gorilla/mux v1.8.0
 	github.com/gorilla/websocket v1.4.2
 	github.com/montanaflynn/stats v0.6.6
 	github.com/prometheus/client_golang v1.11.0

src/go.sum

@@ -43,6 +43,8 @@ github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
 github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI=
+github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
 github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4=

src/load-test-client/main.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/csv"
 	"flag"
+	"github.com/google/uuid"
 	"log"
 	"os"
 	"os/signal"
@@ -26,20 +27,26 @@ func main() {
 	msgSize := flag.Int("msg-size", 256,
 		"The size of the messages in bytes (just for load test mode)")
 
-	numOfClients := flag.Int("clients", 1,
-		"Number of clients that will be started (just for load test mode")
+	numOfRooms := flag.Int("rooms", 1,
+		"Number of chat rooms that will be initialized (just for load test mode)")
+
+	roomSize := flag.Int("room-size", 1,
+		"Number of clients that will be started per room (just for load test mode")
 
 	flag.Parse()
 
 	var msgEvents chan *client.MessageEventEntry
 	var cancelFuncs []*context.CancelFunc
 	waitGroup := &sync.WaitGroup{}
+	numOfClients := *numOfRooms * *roomSize
 
 	if !*loadTest {
-		*numOfClients = 1
+		*numOfRooms = 1
+		*roomSize = 1
+		numOfClients = 1
 	}
 
-	waitGroup.Add(*numOfClients)
+	waitGroup.Add(numOfClients)
 
 	var csvWriterCancelFunc *context.CancelFunc
 	csvWriterWaitGroup := &sync.WaitGroup{}
@@ -54,34 +61,40 @@ func main() {
 		csvWriterWaitGroup.Add(1)
 	}
 
-	// Create numOfClients clients that can chat
-	for i := 1; i <= *numOfClients; i++ {
-		log.Printf("Creating client number: %v / %v", i, *numOfClients)
-
-		// Listen to system interrupts -> program will be stopped
-		closeConnection := make(chan os.Signal, 1)
-		signal.Notify(closeConnection, os.Interrupt)
-
-		ctx, cancelFunc := context.WithCancel(context.Background())
-		cancelFuncs = append(cancelFuncs, &cancelFunc)
-
-		go func() {
-			chatClient := client.Client{
-				Context:          ctx,
-				WaitGroup:        waitGroup,
-				ServerUrl:        *serverUrl,
-				CloseConnection:  closeConnection,
-				IsLoadTestClient: *loadTest,
-				MsgFrequency:     *msgFrequency,
-				MsgSize:          *msgSize,
-				MsgEvents:        msgEvents,
-			}
-
-			err := chatClient.Start()
-			if err != nil {
-				log.Fatalf("%v", err)
-			}
-		}()
+	// Create rooms
+	for i := 0; i < *numOfRooms; i++ {
+		room := uuid.New().String()
+
+		// Create clients per room
+		for j := 1; j <= *roomSize; j++ {
+			log.Printf("Creating client number: %v / %v", j, *roomSize)
+
+			// Listen to system interrupts -> program will be stopped
+			closeConnection := make(chan os.Signal, 1)
+			signal.Notify(closeConnection, os.Interrupt)
+
+			ctx, cancelFunc := context.WithCancel(context.Background())
+			cancelFuncs = append(cancelFuncs, &cancelFunc)
+
+			go func() {
+				chatClient := client.Client{
+					Context:          ctx,
+					WaitGroup:        waitGroup,
+					ServerUrl:        *serverUrl,
+					CloseConnection:  closeConnection,
+					IsLoadTestClient: *loadTest,
+					MsgFrequency:     *msgFrequency,
+					MsgSize:          *msgSize,
+					MsgEvents:        msgEvents,
+					Room:             room,
+				}
+
+				err := chatClient.Start()
+				if err != nil {
+					log.Fatalf("%v", err)
+				}
+			}()
+		}
 	}
 
 	// Listen to system interrupts -> program will be stopped

src/server/client.go

@@ -12,11 +12,13 @@ type Client struct {
 	wsConn    *websocket.Conn
 	outgoing  chan *MessageWrapper
 	waitGroup *sync.WaitGroup
+	room      string
 }
 
 type MessageWrapper struct {
 	message         *chat.Message
 	processingTimer *prometheus.Timer
+	room            string
 }
 
 //HandleOutgoing sends outgoing messages to the client's websocket connection
@@ -69,7 +71,7 @@ func (client *Client) HandleIncoming(incoming chan<- *MessageWrapper) {
 			continue
 		}
 
-		wrapper := MessageWrapper{message: &message, processingTimer: timer}
+		wrapper := MessageWrapper{message: &message, processingTimer: timer, room: client.room}
 
 		incoming <- &wrapper
 	}

src/server/clientsManager.go

@@ -21,7 +21,7 @@ var incoming = make(chan *MessageWrapper, messageBufferSize)
 
 // StartClient starts a client's incoming and outgoing message handlers
 // and waits until the connection breaks to remove the client
-func StartClient(wsConn *websocket.Conn) {
+func StartClient(wsConn *websocket.Conn, room string) {
 	outgoing := make(chan *MessageWrapper, messageBufferSize)
 
 	waitGroup := sync.WaitGroup{}
@@ -31,6 +31,7 @@ func StartClient(wsConn *websocket.Conn) {
 		wsConn:    wsConn,
 		outgoing:  outgoing,
 		waitGroup: &waitGroup,
+		room:      room,
 	}
 	clients = append(clients, &client)
 
@@ -56,6 +57,10 @@ func BroadcastMessages() {
 		chatHistory = append(chatHistory, wrapper.message)
 
 		for _, client := range clients {
+			if wrapper.room != client.room {
+				continue
+			}
+
 			// By providing a default case, we avoid blocking the main broadcasting loop
 			// in case the buffer of the outgoing channel is full.
 			select {

src/server/demo.html

@@ -15,13 +15,14 @@
                 return
             }
 
-            const socket = new WebSocket(`ws://${document.location.host}/ws`)
+            const socket = new WebSocket(`ws://${document.location.host}/ws/demo`)
             const userIdInput = document.getElementById('userIdInput')
             const messageInput = document.getElementById('messageInput')
 
             // RECEIVING MESSAGES
             socket.onopen = function (event) {
                 displayStatusMessage('Connection established!', true)
+                displayStatusMessage('Automatically connected to the `demo` chat room.', true)
             }
 
             socket.onclose = function (event) {

src/server/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"github.com/gorilla/mux"
 	"github.com/gorilla/websocket"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"log"
@@ -18,12 +19,13 @@ func main() {
 	go BroadcastMessages()
 
 	// Register separate ServeMux instances for public endpoints and internal metrics
-	publicMux := http.NewServeMux()
+	publicMux := mux.NewRouter()
 	internalMux := http.NewServeMux()
 
 	// Register public endpoints
 	publicMux.HandleFunc("/", demoHandler)
 	publicMux.HandleFunc("/ws", wsHandler)
+	publicMux.HandleFunc("/ws/{room}", wsHandler)
 
 	// Register Prometheus endpoint
 	internalMux.Handle("/metrics", promhttp.Handler())
@@ -60,13 +62,19 @@ func main() {
 
 // Event handler for the /ws endpoint
 func wsHandler(writer http.ResponseWriter, req *http.Request) {
+
+	log.Println("Got new connection")
+
+	vars := mux.Vars(req)
+	room := vars["room"]
+
 	wsConn, err := upgrader.Upgrade(writer, req, nil)
 	if err != nil {
 		log.Print("Cannot upgrade to websocket connection:", err)
 		return
 	}
 
-	StartClient(wsConn)
+	StartClient(wsConn, room)
 }
 
 // Handles the / endpoint and serves the demo html chat client