Function TestPullSubscribeFetchBatch

test/js_test.go:1589–2020

(t *testing.T)

}

func TestPullSubscribeFetchBatch(t *testing.T) {
	s := RunBasicJetStreamServer()
	defer shutdownJSServerAndRemoveStorage(t, s)

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

	_, err := js.AddStream(&nats.StreamConfig{
		Name:     "TEST",
		Subjects: []string{"foo"},
	})
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	t.Run("basic fetch", func(t *testing.T) {
		defer js.PurgeStream("TEST")
		sub, err := js.PullSubscribe("foo", "")
		if err != nil {
			t.Fatalf("Unexpected error: %s", err)
		}
		for i := 0; i < 5; i++ {
			if _, err := js.Publish("foo", []byte("msg")); err != nil {
				t.Fatalf("Unexpected error: %s", err)
			}
		}
		res, err := sub.FetchBatch(10)
		if err != nil {
			t.Fatalf("Unexpected error: %s", err)
		}
		go func() {
			time.Sleep(10 * time.Millisecond)
			for i := 0; i < 5; i++ {
				js.Publish("foo", []byte("msg"))
			}
		}()
		msgs := make([]*nats.Msg, 0)
		for msg := range res.Messages() {
			msgs = append(msgs, msg)
		}
		if res.Error() != nil {
			t.Fatalf("Unexpected error: %s", res.Error())
		}
		if len(msgs) != 10 {
			t.Fatalf("Expected %d messages; got: %d", 10, len(msgs))
		}
	})

	t.Run("multiple concurrent fetches", func(t *testing.T) {
		defer js.PurgeStream("TEST")
		sub, err := js.PullSubscribe("foo", "")
		if err != nil {
			t.Fatalf("Unexpected error: %s", err)
		}
		for i := 0; i < 50; i++ {
			if _, err := js.Publish("foo", []byte("msg")); err != nil {
				t.Fatalf("Unexpected error: %s", err)
			}