Function TestWebSocketManySubscribers

websocket_test.go:1027–1133

===== SCALABILITY / STRESS TESTS =====

(t *testing.T)

// ===== SCALABILITY / STRESS TESTS =====

func TestWebSocketManySubscribers(t *testing.T) {
	ensureServer()
	token := signUpAndGetToken(t)

	// keep under CI connection limits (2-4 core runners)
	const numSubscribers = 20

	topicName := fmt.Sprintf("fan-out-%d", time.Now().UnixNano())

	// create topic
	wsCreator := dialWS(t, token)
	sendJSON(t, wsCreator, wsPayload{Id: nextReqId(), Method: "create-topicName", Payload: map[string]interface{}{"name": topicName}})
	recvJSON(t, wsCreator, 5*time.Second) // create response
	time.Sleep(300 * time.Millisecond)

	// connect and subscribe N clients
	subscribers := make([]*websocket.Conn, numSubscribers)
	var subWg sync.WaitGroup
	for i := 0; i < numSubscribers; i++ {
		subWg.Add(1)
		go func(idx int) {
			defer subWg.Done()
			config, err := websocket.NewConfig(wsURL, wsBaseAddress)
			if err != nil {
				t.Errorf("subscriber %d: config error: %v", idx, err)
				return
			}
			config.Header.Set("Authorization", "Bearer "+token)
			config.Header.Set("Cookie", "token="+token)
			ws, err := websocket.DialConfig(config)
			if err != nil {
				return
			}
			// consume session-open
			ws.SetReadDeadline(time.Now().Add(5 * time.Second))
			var sessionMsg resource.WsOutMessage
			if err := websocket.JSON.Receive(ws, &sessionMsg); err != nil {
				ws.Close()
				return
			}
			if err := websocket.JSON.Send(ws, wsPayload{Id: nextReqId(), Method: "subscribe", Payload: map[string]interface{}{"topicName": topicName}}); err != nil {
				ws.Close()
				return
			}
			ws.SetReadDeadline(time.Now().Add(30 * time.Second))
			var ack resource.WsOutMessage
			if err := websocket.JSON.Receive(ws, &ack); err != nil {
				ws.Close()
				return
			}
			subscribers[idx] = ws
		}(i)
	}
	subWg.Wait()

	// count how many actually connected
	var connectedCount int
	for _, ws := range subscribers {