Function TestNoRaceJetStreamConsumerSlowConsumer

test/norace_test.go:106–174

(t *testing.T)

}

func TestNoRaceJetStreamConsumerSlowConsumer(t *testing.T) {
	// This test fails many times, need to look harder at the imbalance.
	t.SkipNow()

	s := RunServerOnPort(-1)
	defer shutdownJSServerAndRemoveStorage(t, s)

	if err := s.EnableJetStream(nil); err != nil {
		t.Fatalf("Expected no error, got %v", err)
	}

	nc, js := jsClient(t, s)
	defer nc.Close()

	var err error

	_, err = js.AddStream(&nats.StreamConfig{
		Name:     "PENDING_TEST",
		Subjects: []string{"js.p"},
		Storage:  nats.MemoryStorage,
	})
	if err != nil {
		t.Fatalf("stream create failed: %v", err)
	}

	// Override default handler for test.
	nc.SetErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, _ error) {})

	// Queue up 1M small messages.
	toSend := uint64(1000000)
	for i := uint64(0); i < toSend; i++ {
		nc.Publish("js.p", []byte("ok"))
	}
	nc.Flush()

	str, err := js.StreamInfo("PENDING_TEST")
	if err != nil {
		t.Fatal(err)
	}

	if nm := str.State.Msgs; nm != toSend {
		t.Fatalf("Expected to have stored all %d msgs, got only %d", toSend, nm)
	}

	var received uint64
	done := make(chan bool, 1)

	js.Subscribe("js.p", func(m *nats.Msg) {
		received++
		if received >= toSend {
			done <- true
		}
		meta, err := m.Metadata()
		if err != nil {
			t.Fatalf("could not get message metadata: %s", err)
		}
		if meta.Sequence.Stream != received {
			t.Errorf("Missed a sequence, was expecting %d but got %d, last error: '%v'", received, meta.Sequence.Stream, nc.LastError())