bench single conn many events

controller.go

@@ -192,9 +192,10 @@ func NewController(name string, options ...ControllerOption) Controller {
 
 	o := &opt{
 		websocketUpgrader: websocket.Upgrader{
-			EnableCompression: true,
-			// ReadBufferSize:    256,
-			// WriteBufferSize:   256,
+			// disabled compression since its too noisy: https://github.com/gorilla/websocket/issues/859
+			// EnableCompression: true,
+			// ReadBufferSize:  4096,
+			// WriteBufferSize: 4096,
 			// WriteBufferPool: &sync.Pool{},
 		},
 		watchExts:   defaultWatchExtensions,

controller_test.go

@@ -63,6 +63,9 @@ type testInput struct {
 	num       int
 	wssend    int
 	wsrecv    int
+	event     Event
+	conn      *websocket.Conn
+	closeWs   bool
 	sync.RWMutex
 }
 
@@ -121,7 +124,7 @@ func eventPayload(tb testing.TB, ti *testInput) Event {
 		IsForm:    false,
 		Params:    data,
 		SessionID: &firSession,
-		Timestamp: time.Now().UnixNano(),
+		Timestamp: time.Now().UTC().UnixMilli(),
 	}
 
 	return event
@@ -180,35 +183,49 @@ func runPostEventTest(tb testing.TB, ti *testInput) {
 
 }
 
-func runWebsocketEventTest(tb testing.TB, ti *testInput) {
-
-	event := eventPayload(tb, ti)
-
-	// same as above but with websocket
+func dialWebSocket(tb testing.TB, ti *testInput, event Event) *websocket.Conn {
 	wsURLString := strings.Replace(ti.serverURL, "http", "ws", 1)
 	wsDialer := websocket.Dialer{
 		Proxy:            http.ProxyFromEnvironment,
 		HandshakeTimeout: 45 * time.Second,
-		// Jar:              jar,
+		// EnableCompression: true,
+		// Jar:              jar, // jar doesnät work but adding a Cookie header does
 	}
 	header := http.Header{}
 	header.Set("Cookie", fmt.Sprintf("_fir_session_=%s", *event.SessionID))
 	ws, _, err := wsDialer.Dial(wsURLString, header)
 	if err != nil {
 		tb.Fatal(err)
 	}
-	defer ws.Close()
+
+	return ws
+}
+
+func runWebsocketEventTest(tb testing.TB, ti *testInput) {
+
+	event := ti.event
+	if event.SessionID == nil {
+		event = eventPayload(tb, ti)
+	}
+
+	ws := ti.conn
+	if ws == nil {
+		ws = dialWebSocket(tb, ti, event)
+	}
+
+	if ti.closeWs {
+		defer ws.Close()
+	}
 
 	ti.incWssend()
 
-	err = ws.WriteJSON(event)
+	err := ws.WriteJSON(event)
 	if err != nil {
 		tb.Fatal(err)
 	}
 
-	ws.SetReadDeadline(time.Now().Add(10000 * time.Millisecond))
+	ws.SetReadDeadline(time.Now().Add(1000 * time.Millisecond))
 	var message []byte
-
 	_, message, err = ws.ReadMessage()
 	if err != nil {
 		tb.Fatal(err)
@@ -230,6 +247,10 @@ func runWebsocketEventTest(tb testing.TB, ti *testInput) {
 		tb.Fatalf("expected: %s, got: %s", expectedHTML, domEvents[0].Detail.HTML)
 	}
 
+	if !ti.closeWs {
+		return
+	}
+
 	err = ws.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""))
 	if err != nil {
 		tb.Fatal(err)
@@ -278,7 +299,7 @@ func BenchmarkControllerWebsocktEnabled(b *testing.B) {
 		server := httptest.NewServer(controller.RouteFunc(tc.routeFunc))
 		defer server.Close()
 
-		ti := &testInput{serverURL: server.URL, num: 10}
+		ti := &testInput{serverURL: server.URL, num: 10, closeWs: true}
 		b.Cleanup(func() {
 			fmt.Printf("ws send: %d, ws recv: %d\n", ti.getWssend(), ti.getWsrecv())
 		})
@@ -291,7 +312,40 @@ func BenchmarkControllerWebsocktEnabled(b *testing.B) {
 		b.ReportMetric(float64(ti.getWsrecv()), "total_receives")
 		b.ReportAllocs()
 	}
+}
 
+func TestControllerWebsocktEnabledMultiEvent(t *testing.T) {
+	for _, tc := range testCases {
+		controller := NewController(tc.name, tc.options...)
+		// Create a test HTTP server
+		server := httptest.NewServer(controller.RouteFunc(tc.routeFunc))
+		defer server.Close()
+
+		ti := &testInput{serverURL: server.URL, num: 10}
+		ti.event = eventPayload(t, ti)
+		ti.conn = dialWebSocket(t, ti, ti.event)
+
+		t.Cleanup(func() {
+			err := ti.conn.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""))
+			if err != nil {
+				t.Fatal(err)
+			}
+			time.Sleep(500 * time.Millisecond)
+			ti.conn.Close()
+
+		})
+
+		for i := 0; i < 20; i++ {
+			time.Sleep(251 * time.Millisecond)
+			ti.event.Timestamp = time.Now().UTC().UnixMilli()
+			runWebsocketEventTest(t, ti)
+		}
+
+		if ti.getWssend() != ti.getWsrecv() {
+			t.Fatalf("expected ws send: %d, ws recv: %d", ti.getWssend(), ti.getWsrecv())
+		}
+
+	}
 }
 
 func TestControllerWebsocketEnabled(t *testing.T) {

websocket.go

@@ -20,7 +20,7 @@ import (
 
 const (
 	// Time allowed to write a message to the peer.
-	writeWait = 10 * time.Second
+	writeWait = 20 * time.Second
 
 	// Time allowed to read the next pong message from the peer.
 	pongWait = 55 * time.Second
@@ -184,16 +184,24 @@ func onWebsocket(w http.ResponseWriter, r *http.Request, cntrl *controller) {
 	}
 
 	conn.SetReadLimit(maxMessageSize)
-	conn.EnableWriteCompression(true)
-	conn.SetCompressionLevel(9)
-	conn.SetWriteDeadline(time.Now().Add(writeWait))
+	// disabled compression since its too noisy: https://github.com/gorilla/websocket/issues/859
+	//conn.EnableWriteCompression(true)
+	//conn.SetCompressionLevel(4)
 	conn.SetReadDeadline(time.Now().Add(pongWait))
 	conn.SetPongHandler(func(string) error {
 		//logger.Infof("pong from %v", conn.RemoteAddr())
 		conn.SetReadDeadline(time.Now().Add(pongWait))
 		return nil
 	})
 
+	// workaround for noisy logs due to close handler printing a useless error
+	//  https://github.com/gorilla/websocket/issues/880
+	conn.SetCloseHandler(func(code int, text string) error {
+		message := websocket.FormatCloseMessage(code, "")
+		conn.WriteControl(websocket.CloseMessage, message, time.Now().Add(writeWait))
+		return nil
+	})
+
 	writePumpDone := make(chan struct{})
 	go writePump(conn, writePumpDone, send)
 
@@ -208,10 +216,10 @@ loop:
 
 		_, message, err := conn.ReadMessage()
 		if err != nil {
-
 			if !websocket.IsCloseError(err, websocket.CloseNormalClosure) && websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
 				logger.Errorf("read: %v, %v", conn.RemoteAddr().String(), err)
 			}
+
 			break loop
 		}
 		// start := time.Now()
@@ -333,7 +341,6 @@ func writeEvent(send chan []byte, pubsubEvent pubsub.Event) error {
 }
 
 func writeConn(conn *websocket.Conn, mt int, payload []byte) error {
-	conn.SetWriteDeadline(time.Now().Add(writeWait))
 	return conn.WriteMessage(mt, payload)
 }
 
@@ -347,6 +354,7 @@ loop:
 	for {
 		select {
 		case message, ok := <-send:
+			conn.SetWriteDeadline(time.Now().Add(writeWait))
 			if !ok {
 				err := writeConn(conn, websocket.CloseMessage, []byte{})
 				if err != nil {
@@ -368,14 +376,17 @@ loop:
 			}
 
 			if err := w.Close(); err != nil {
+				logger.Errorf("close err: %v", err)
 				return
 			}
 
 		case <-closeWritePump:
 			break loop
 		case <-ticker.C:
 			//logger.Infof("ping to client: %v", conn.RemoteAddr())
+			conn.SetWriteDeadline(time.Now().Add(writeWait))
 			if err := writeConn(conn, websocket.PingMessage, []byte{}); err != nil {
+				logger.Errorf("ping err: %v", err)
 				return
 			}
 		}