Function TestPullSubscribeConsumerDoesNotExist

test/js_test.go:1339–1388

(t *testing.T)

}

func TestPullSubscribeConsumerDoesNotExist(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)
	}

	sub, err := js.PullSubscribe("foo", "")
	if err != nil {
		t.Fatalf("Unexpected error: %s", err)
	}
	defer sub.Unsubscribe()

	info, err := sub.ConsumerInfo()
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	if err := js.DeleteConsumer("TEST", info.Name); err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	_, err = sub.Fetch(5)
	if !errors.Is(err, nats.ErrNoResponders) {
		t.Fatalf("Expected no responders error; got: %v", err)
	}

	msgs, err := sub.FetchBatch(5)
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	select {
	case _, ok := <-msgs.Messages():
		if ok {
			t.Fatalf("Expected no messages")
		}
	case <-time.After(time.Second):
		t.Fatalf("Timeout waiting for messages")
	}

	if !errors.Is(msgs.Error(), nats.ErrNoResponders) {
		t.Fatalf("Expected no responders error; got: %v", msgs.Error())
	}
}

func TestPullSubscribeFetchDrain(t *testing.T) {
	s := RunBasicJetStreamServer()