Function TestWebSocketConcurrentPublishers

websocket_test.go:1423–1495

(t *testing.T)

}

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

	const numPublishers = 5
	const msgsPerPublisher = 50

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

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

	// subscriber
	wsSub := dialWS(t, token)
	defer wsSub.Close()
	sendJSON(t, wsSub, wsPayload{Id: nextReqId(), Method: "subscribe", Payload: map[string]interface{}{"topicName": topicName}})
	recvJSON(t, wsSub, 5*time.Second) // subscribe ack

	// launch concurrent publishers
	var wg sync.WaitGroup
	var sendErrors int64
	start := time.Now()
	for p := 0; p < numPublishers; p++ {
		wg.Add(1)
		go func(pubIdx int) {
			defer wg.Done()
			ws := dialWS(t, token)
			defer ws.Close()
			for m := 0; m < msgsPerPublisher; m++ {
				err := websocket.JSON.Send(ws, wsPayload{
					Method: "new-message",
					Payload: map[string]interface{}{
						"topicName": topicName,
						"message":   map[string]interface{}{"pub": pubIdx, "seq": m},
					},
				})
				if err != nil {
					atomic.AddInt64(&sendErrors, 1)
					return
				}
			}
		}(p)
	}
	wg.Wait()
	sendDuration := time.Since(start)
	totalSent := int64(numPublishers*msgsPerPublisher) - sendErrors
	t.Logf("sent %d messages from %d publishers in %v (%.0f msg/s), %d send errors",
		totalSent, numPublishers, sendDuration, float64(totalSent)/sendDuration.Seconds(), sendErrors)

	// receive what we can
	var recvCount int64
	recvStart := time.Now()
	for recvCount < totalSent {
		_, ok := tryRecvJSON(wsSub, 5*time.Second)
		if !ok {