Implementation of hub client handling

Author: stefaluc

app/src/App.js

@@ -6,6 +6,7 @@ import './App.css';
 const socket = new WebSocket("ws://127.0.0.1:8081/ws");
 socket.onopen = (event) => {
   console.log('Opened socket');
+  console.log(event);
 };
 
 socket.onerror = (event) => {

server/conn_hub.go

@@ -1 +1,46 @@
 package main
+
+type ConnHub struct {
+	clients map[*Client]bool
+	broadcast chan []byte
+	register chan *Client
+	unregister chan *Client
+}
+
+// init ConnHub
+func newConnHub() *ConnHub {
+	return &ConnHub{
+		clients: make(map[*Client]bool),
+		broadcast: make(chan []byte),
+		register: make(chan *Client),
+		unregister: make(chan *Client),
+	}
+}
+
+// handles channel actions of a ConnHub
+func (hub *ConnHub) run() {
+	for {
+		select {
+		// register client to hub
+		case client := <-hub.register:
+			hub.clients[client] = true
+		// unregister client to hub
+		case client := <-hub.unregister:
+			if _, ok := hub.clients[client]; ok {
+				delete(hub.clients, client)
+				close(client.send)
+			}
+		// loop through registered clients and send message to their send channel
+		case message := <-hub.broadcast:
+			for client := range hub.clients {
+				select {
+				case client.send <- message:
+				// if send buffer is full, assume client is dead or stuck and unregister
+				default:
+					close(client.send)
+					delete(hub.clients, client)
+				}
+			}
+		}
+	}
+}

server/main.go

@@ -9,11 +9,11 @@ func main() {
 	fmt.Println("Launching server...")
 
 	// start conn_hub
-	// connHub := newConnHub
-	// go connHub.run()
+	hub := newConnHub()
+	go hub.run()
 
 	http.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) {
-		wsHandler(/* hub ,*/ w, r)
+		wsHandler(hub, w, r)
 	})
 	err := http.ListenAndServe(":8081", nil)
 	if err != nil {

server/websocket.go

@@ -1,16 +1,23 @@
 package main
 
 import (
-	"fmt"
+	// "fmt"
 	"net/http"
+	"time"
+
 	"github.com/gorilla/websocket"
 )
 
-type JsonData struct {
-	Text string `json:"text"`
-	Client string `json:"client"`
-	Timestamp string `json:"timestamp"`
-}
+const (
+	// Time allowed to write a message to the peer.
+	writeWait = 10 * time.Second
+	// Time allowed to read the next pong message from the peer.
+	pongWait = 60 * time.Second
+	// Send pings to peer with this period. Must be less than pongWait.
+	pingPeriod = (pongWait * 9) / 10
+	// Maximum message size allowed from peer.
+	maxMessageSize = 512
+)
 
 var upgrader = websocket.Upgrader{
 	ReadBufferSize: 1024,
@@ -19,32 +26,105 @@ var upgrader = websocket.Upgrader{
 	CheckOrigin: func(r *http.Request) bool { return true },
 }
 
-// handle /ws route, upgrade HTTP request and forward ws conn to worker
-func wsHandler(w http.ResponseWriter, r *http.Request) {
-	conn, err := upgrader.Upgrade(w, r, nil)
-	if err != nil {
-		http.Error(w, "Could not open websocket connection", http.StatusBadRequest)
-	}
+type JsonData struct {
+	Text string `json:"text"`
+	Client string `json:"client"`
+	Timestamp string `json:"timestamp"`
+}
 
-	go handleConnection(conn)
+type Client struct {
+	hub *ConnHub
+	conn *websocket.Conn
+	send chan []byte
 }
 
-func handleConnection(conn *websocket.Conn) {
-	fmt.Println("Handling new connection")
+func (c *Client) readPump() {
+	// schedule client to be disconnected
+	defer func() {
+		c.hub.unregister <- c
+		c.conn.Close()
+	}()
 
+	// init client connection
+	c.conn.SetReadLimit(maxMessageSize)
+	c.conn.SetReadDeadline(time.Now().Add(pongWait))
+	c.conn.SetPongHandler(func(string) error {
+		c.conn.SetReadDeadline(time.Now().Add(pongWait))
+		return nil
+	})
+
+	// handle connection read
 	for {
-		fmt.Println("Awaiting data from connection...")
+		// read JSON data from connection
+		// message := JsonData{}
+		// if err := c.conn.ReadJSON(&message); err != nil {
+		// 	fmt.Println("Error reading JSON", err)
+		// }
+		// fmt.Printf("Get response: %#v\n", message)
 
-		res := JsonData{}
+		// // queue message for writing
+		// c.hub.broadcast <- message
+	}
+}
 
-		if err := conn.ReadJSON(&res); err != nil {
-			fmt.Println("Error reading JSON", err)
-		}
+func (c *Client) writePump() {
+	ticker := time.NewTicker(pingPeriod)
+	defer func() {
+		ticker.Stop()
+		c.conn.Close()
+	}()
 
-		fmt.Printf("Get response: %#v\n", res)
+	for {
+		select {
+		case message, ok := <-c.send:
+			c.conn.SetWriteDeadline(time.Now().Add(writeWait))
+			if !ok {
+				// channel has been closed by the hub
+				c.conn.WriteMessage(websocket.CloseMessage, []byte{})
+				return
+			}
+
+			w, err := c.conn.NextWriter(websocket.TextMessage)
+			if err != nil {
+				return
+			}
+			w.Write(message)
 
-		if err := conn.WriteJSON(res); err != nil {
-			fmt.Println("Error writing JSON", err)
+			// coalesce pending messages into one message
+			n := len(c.send)
+			for i := 0; i < n; i++ {
+				w.Write(<-c.send)
+			}
+
+			if err := w.Close(); err != nil {
+				return
+			}
+		// send ping over websocket
+		case <-ticker.C:
+			c.conn.SetWriteDeadline(time.Now().Add(writeWait))
+			if err := c.conn.WriteMessage(websocket.PingMessage, []byte{}); err != nil {
+				return
+			}
 		}
 	}
 }
+
+// handle /ws route, upgrade HTTP request and begin handling of client conn
+func wsHandler(hub *ConnHub, w http.ResponseWriter, r *http.Request) {
+	conn, err := upgrader.Upgrade(w, r, nil)
+	if err != nil {
+		http.Error(w, "Could not open websocket connection", http.StatusBadRequest)
+	}
+
+	// init new client, register to hub
+	client := &Client{
+		hub: hub,
+		conn: conn,
+		send: make(chan []byte, 256),
+	}
+	client.hub.register <- client
+
+	// separate reads and writes to conform to WebSocket standard of one concurrent reader and writer
+	// go client.writePump()
+	// go client.readPump()
+}