feat(server): initial WebSocket hub and client

Author: aboqasem

server/.env.example

@@ -1,2 +1,3 @@
 PORT=8080
+CORS_ORIGIN=http://localhost:5173
 LOG_LEVEL=debug

server/config/config.go

@@ -9,19 +9,29 @@ import (
 )
 
 var logger = log.With().Str("pkg", "config").Logger()
+var conf *Config
 
 type Config struct {
-	Port     uint16
-	LogLevel zerolog.Level
+	Port       uint16
+	CorsOrigin string
+	LogLevel   zerolog.Level
 }
 
-func Init() *Config {
+func new() *Config {
 	return &Config{
-		Port:     parseEnv("PORT", 8080, parseUint16),
-		LogLevel: parseEnv("LOG_LEVEL", zerolog.InfoLevel, parseLogLevel),
+		Port:       parseEnv("PORT", 8080, parseUint16),
+		CorsOrigin: parseEnv("CORS_ORIGIN", "*", nop),
+		LogLevel:   parseEnv("LOG_LEVEL", zerolog.InfoLevel, parseLogLevel),
 	}
 }
 
+func Get() *Config {
+	if conf == nil {
+		conf = new()
+	}
+	return conf
+}
+
 // UTILS
 
 func parseEnv[V any, P func(string) (V, error)](key string, defaultVal V, parser P) V {
@@ -38,6 +48,10 @@ func parseEnv[V any, P func(string) (V, error)](key string, defaultVal V, parser
 
 // PARSERS
 
+func nop(s string) (string, error) {
+	return s, nil
+}
+
 func parseUint(s string, bits int) (uint64, error) {
 	return strconv.ParseUint(s, 10, bits)
 }

server/go.mod

@@ -4,7 +4,10 @@ go 1.21.7
 
 require github.com/rs/zerolog v1.31.0
 
+require golang.org/x/net v0.17.0 // indirect
+
 require (
+	github.com/gorilla/websocket v1.5.1
 	github.com/joho/godotenv v1.5.1
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.19 // indirect

server/go.sum

@@ -1,5 +1,7 @@
 github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
 github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
+github.com/gorilla/websocket v1.5.1 h1:gmztn0JnHVt9JZquRuzLw3g4wouNVzKL15iLr/zn/QY=
+github.com/gorilla/websocket v1.5.1/go.mod h1:x3kM2JMyaluk02fnUJpQuwD2dCS5NDG2ZHL0uE0tcaY=
 github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0=
 github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4=
 github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA=
@@ -11,6 +13,8 @@ github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE
 github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
 github.com/rs/zerolog v1.31.0 h1:FcTR3NnLWW+NnTwwhFWiJSZr4ECLpqCm6QsEnyvbV4A=
 github.com/rs/zerolog v1.31.0/go.mod h1:/7mN4D5sKwJLZQ2b/znpjC3/GQWY/xaDXUM0kKWRHss=
+golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM=
+golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=

server/main.go

@@ -2,42 +2,37 @@ package main
 
 import (
 	"fmt"
-	"html"
 	"net/http"
 	"time"
 
 	"github.com/aboqasem/portfolio/server/config"
+	"github.com/aboqasem/portfolio/server/ws"
 	_ "github.com/joho/godotenv/autoload"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 )
 
 var logger = log.With().Str("pkg", "main").Logger()
-var conf *config.Config
+var conf = config.Get()
 
 func init() {
-	conf = config.Init()
-
 	zerolog.SetGlobalLevel(conf.LogLevel)
 }
 
 func main() {
 	http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
-		logger.Debug().Any("method", r.Method).Str("path", r.URL.Path).Send()
-
-		w.Write([]byte("<h1>Hello, world!</h1><pre>"))
-		w.Write([]byte(r.Method))
-		w.Write([]byte{' '})
-		w.Write([]byte(html.EscapeString(r.URL.Path)))
-		w.Write([]byte("</pre>"))
+		w.Header().Add("Access-Control-Allow-Origin", conf.CorsOrigin)
+		w.Write([]byte("OK"))
 	})
 
+	http.HandleFunc("/ws/hub", ws.ServeWsHub)
+
 	server := &http.Server{
 		Addr:              fmt.Sprintf(":%d", conf.Port),
 		ReadHeaderTimeout: 3 * time.Second,
 	}
 
-	logger.Info().Uint16("port", conf.Port).Msg("Running server...")
+	logger.Info().Msgf("Running server on http://localhost:%d", conf.Port)
 	err := server.ListenAndServe()
 	if err != nil {
 		logger.Fatal().Err(err).Msg("ListenAndServe")

server/ws/client.go

@@ -0,0 +1,129 @@
+package ws
+
+import (
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+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 = 1_024
+)
+
+var (
+	arrayStart = []byte{'['}
+	arrayEnd   = []byte{']'}
+	comma      = []byte{','}
+)
+
+// Client is a middleman between the websocket connection and the hub.
+type Client struct {
+	hub *Hub
+
+	// The websocket connection.
+	conn *websocket.Conn
+
+	// Buffered channel of outbound messages.
+	send chan []byte
+}
+
+// readPump pumps messages from the websocket connection to the hub.
+//
+// The application runs readPump in a per-connection goroutine. The application
+// ensures that there is at most one reader on a connection by executing all
+// reads from this goroutine.
+func (c *Client) readPump() {
+	defer func() {
+		c.hub.unregister <- c
+		c.conn.Close()
+	}()
+
+	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
+	})
+
+	for {
+		_, message, err := c.conn.ReadMessage()
+		if err != nil {
+			if websocket.IsUnexpectedCloseError(
+				err,
+				websocket.CloseGoingAway,
+				websocket.CloseAbnormalClosure,
+				websocket.CloseNoStatusReceived,
+			) {
+				logger.Error().Err(err).Msg("unexpected close error")
+			}
+			break
+		}
+
+		c.hub.broadcast <- message
+	}
+}
+
+// writePump pumps messages from the hub to the websocket connection.
+//
+// A goroutine running writePump is started for each connection. The
+// application ensures that there is at most one writer to a connection by
+// executing all writes from this goroutine.
+func (c *Client) writePump() {
+	ticker := time.NewTicker(pingPeriod)
+	defer func() {
+		ticker.Stop()
+		c.conn.Close()
+	}()
+
+	for {
+		select {
+		case message, ok := <-c.send:
+			c.conn.SetWriteDeadline(time.Now().Add(writeWait))
+			if !ok {
+				// The hub closed the channel.
+				c.conn.WriteMessage(websocket.CloseMessage, []byte{})
+				return
+			}
+
+			w, err := c.conn.NextWriter(websocket.TextMessage)
+			if err != nil {
+				return
+			}
+
+			n := len(c.send)
+			if n == 0 {
+				w.Write(message)
+			} else {
+				w.Write(arrayStart)
+				w.Write(message)
+				w.Write(comma)
+				for i := 1; i < n; i++ {
+					w.Write(<-c.send)
+					w.Write(comma)
+				}
+				w.Write(<-c.send)
+				w.Write(arrayEnd)
+			}
+
+			if err := w.Close(); err != nil {
+				return
+			}
+
+		case <-ticker.C:
+			c.conn.SetWriteDeadline(time.Now().Add(writeWait))
+			if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
+				return
+			}
+		}
+	}
+}

server/ws/hub.go

@@ -0,0 +1,51 @@
+package ws
+
+// Hub maintains the set of active clients and broadcasts messages to the
+// clients.
+type Hub struct {
+	// Registered clients.
+	clients map[*Client]bool
+
+	// Inbound messages from the clients.
+	broadcast chan []byte
+
+	// Register requests from the clients.
+	register chan *Client
+
+	// Unregister requests from clients.
+	unregister chan *Client
+}
+
+func newHub() *Hub {
+	return &Hub{
+		broadcast:  make(chan []byte),
+		register:   make(chan *Client),
+		unregister: make(chan *Client),
+		clients:    make(map[*Client]bool),
+	}
+}
+
+func (h *Hub) run() {
+	for {
+		select {
+		case client := <-h.register:
+			h.clients[client] = true
+
+		case client := <-h.unregister:
+			if _, ok := h.clients[client]; ok {
+				close(client.send)
+				delete(h.clients, client)
+			}
+
+		case message := <-h.broadcast:
+			for client := range h.clients {
+				select {
+				case client.send <- message:
+				default:
+					close(client.send)
+					delete(h.clients, client)
+				}
+			}
+		}
+	}
+}

server/ws/serve.go

@@ -0,0 +1,45 @@
+package ws
+
+import (
+	"net/http"
+
+	"github.com/aboqasem/portfolio/server/config"
+	"github.com/gorilla/websocket"
+	"github.com/rs/zerolog/log"
+)
+
+var (
+	logger = log.With().Str("pkg", "ws").Logger()
+
+	conf = config.Get()
+
+	hub = newHub()
+
+	upgrader = websocket.Upgrader{
+		ReadBufferSize:  1024,
+		WriteBufferSize: 1024,
+		CheckOrigin: func(r *http.Request) bool {
+			return conf.CorsOrigin == "*" || conf.CorsOrigin == r.Header.Get("Origin")
+		},
+	}
+)
+
+func init() {
+	go hub.run()
+}
+
+func ServeWsHub(w http.ResponseWriter, r *http.Request) {
+	conn, err := upgrader.Upgrade(w, r, nil)
+	if err != nil {
+		logger.Error().Err(err).Msg("failed to upgrade connection")
+		return
+	}
+
+	client := &Client{hub: hub, conn: conn, send: make(chan []byte, 256)}
+	client.hub.register <- client
+
+	// Allow collection of memory referenced by the caller by doing all work in new goroutines.
+	go client.writePump()
+	go client.readPump()
+
+}