Function TestJetStreamPublishAsyncPerf

test/js_test.go:8439–8495

(t *testing.T)

}

func TestJetStreamPublishAsyncPerf(t *testing.T) {
	// Comment out below to run this benchmark.
	t.SkipNow()

	s := RunBasicJetStreamServer()
	defer shutdownJSServerAndRemoveStorage(t, s)

	nc, err := nats.Connect(s.ClientURL())
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	defer nc.Close()

	// 64 byte payload.
	msg := make([]byte, 64)
	rand.Read(msg)

	// Setup error handler.
	var errors uint32
	errHandler := func(js nats.JetStream, originalMsg *nats.Msg, err error) {
		t.Logf("Got an async err: %v", err)
		atomic.AddUint32(&errors, 1)
	}

	js, err := nc.JetStream(
		nats.PublishAsyncErrHandler(errHandler),
		nats.PublishAsyncMaxPending(256),
	)
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	if _, err := js.AddStream(&nats.StreamConfig{Name: "B"}); err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	toSend := 1000000
	start := time.Now()
	for i := 0; i < toSend; i++ {
		if _, err = js.PublishAsync("B", msg); err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}
	}

	select {
	case <-js.PublishAsyncComplete():
		if ne := atomic.LoadUint32(&errors); ne > 0 {
			t.Fatalf("Got unexpected errors publishing")
		}
	case <-time.After(5 * time.Second):
		t.Fatalf("Did not receive completion signal")
	}

	tt := time.Since(start)
	fmt.Printf("Took %v to send %d msgs\n", tt, toSend)
	fmt.Printf("%.0f msgs/sec\n\n", float64(toSend)/tt.Seconds())
}