Function TestJetStream_Unsubscribe

test/js_test.go:5679–5814

(t *testing.T)

}

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

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

	var err error

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

	fetchConsumers := func(t *testing.T, expected int) {
		t.Helper()
		checkFor(t, time.Second, 15*time.Millisecond, func() error {
			var infos []*nats.ConsumerInfo
			for info := range js.Consumers("foo") {
				infos = append(infos, info)
			}
			if len(infos) != expected {
				return fmt.Errorf("Expected %d consumers, got: %d", expected, len(infos))
			}
			return nil
		})
	}

	deleteAllConsumers := func(t *testing.T) {
		t.Helper()
		for cn := range js.ConsumerNames("foo") {
			js.DeleteConsumer("foo", cn)
		}
	}

	js.Publish("foo.A", []byte("A"))
	js.Publish("foo.B", []byte("B"))
	js.Publish("foo.C", []byte("C"))

	t.Run("consumers deleted on unsubscribe", func(t *testing.T) {
		sub, err := js.SubscribeSync("foo.A")
		if err != nil {
			t.Fatal(err)
		}
		if err := sub.Unsubscribe(); err != nil {
			t.Errorf("Unexpected error: %v", err)
		}

		sub, err = js.SubscribeSync("foo.B", nats.Durable("B"))
		if err != nil {
			t.Fatal(err)
		}
		if err := sub.Unsubscribe(); err != nil {
			t.Errorf("Unexpected error: %v", err)
		}