Function TestJetStreamDrainFailsToDeleteConsumer

test/js_test.go:9144–9202

(t *testing.T)

}

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

	errCh := make(chan error, 1)
	nc, err := nats.Connect(s.ClientURL(), nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) {
		select {
		case errCh <- err:
		default:
		}
	}))
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	defer nc.Close()

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

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

	js.Publish("foo", []byte("hi"))

	blockCh := make(chan struct{})
	sub, err := js.Subscribe("foo", func(m *nats.Msg) {
		<-blockCh
	}, nats.Durable("dur"))
	if err != nil {
		t.Fatalf("Error subscribing: %v", err)
	}

	// Initiate the drain... it won't complete because we have blocked the
	// message callback.
	sub.Drain()

	// Now delete the JS consumer
	if err := js.DeleteConsumer("TEST", "dur"); err != nil {
		t.Fatalf("Error deleting consumer: %v", err)
	}

	// Now unblock and make sure we get the async error
	close(blockCh)

	select {
	case err := <-errCh:
		if !strings.Contains(err.Error(), "consumer not found") {
			t.Fatalf("Unexpected async error: %v", err)
		}
	case <-time.After(time.Second):
		t.Fatal("Did not get async error")
	}