Function TestJetStreamBindConsumer

test/js_test.go:8785–8965

(t *testing.T)

}

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

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

	if _, err := js.AddStream(nil); err == nil {
		t.Fatalf("Unexpected success")
	}
	si, err := js.AddStream(&nats.StreamConfig{Name: "foo"})
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	if si == nil || si.Config.Name != "foo" {
		t.Fatalf("StreamInfo is not correct %+v", si)
	}

	for i := 0; i < 25; i++ {
		js.Publish("foo", []byte("hi"))
	}

	// Both stream and consumer names are required for bind only.
	_, err = js.SubscribeSync("foo", nats.Bind("", ""))
	if err != nats.ErrStreamNameRequired {
		t.Fatalf("Unexpected error: %v", err)
	}
	_, err = js.SubscribeSync("foo", nats.Bind("foo", ""))
	if err != nats.ErrConsumerNameRequired {
		t.Fatalf("Unexpected error: %v", err)
	}
	_, err = js.SubscribeSync("foo", nats.Bind("foo", "push"))
	if err == nil || !errors.Is(err, nats.ErrConsumerNotFound) {
		t.Fatalf("Unexpected error: %v", err)
	}

	// Pull consumer
	_, err = js.PullSubscribe("foo", "pull", nats.Bind("foo", "pull"))
	if err == nil || !errors.Is(err, nats.ErrConsumerNotFound) {
		t.Fatalf("Unexpected error: %v", err)
	}

	// Push consumer
	_, err = js.AddConsumer("foo", &nats.ConsumerConfig{
		Durable:        "push",
		AckPolicy:      nats.AckExplicitPolicy,
		DeliverSubject: nats.NewInbox(),
	})
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	// Push Consumer Bind Only
	sub, err := js.SubscribeSync("foo", nats.Bind("foo", "push"))
	if err != nil {
		t.Fatal(err)
	}
	// Ambiguous declaration should not be allowed.