Function TestJetStreamSubscribe

test/js_test.go:529–1073

(t *testing.T)

}

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

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

	var err error

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

	// Create the stream using our client API.
	_, err = js.AddStream(&nats.StreamConfig{
		Name:     "TEST",
		Subjects: []string{"foo", "bar", "baz", "foo.*"},
	})
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}

	// Lookup the stream for testing.
	_, err = js.StreamInfo("TEST")
	if err != nil {
		t.Fatalf("stream lookup failed: %v", err)
	}

	// If stream name is not specified, then the subject is required.
	if _, err := js.SubscribeSync(""); err == nil || !strings.Contains(err.Error(), "required") {
		t.Fatalf("Unexpected error: %v", err)
	}
	// Check that if stream name is present, then technically the subject does not have to.
	sub, err := js.SubscribeSync("", nats.BindStream("TEST"))
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	initialPending, err := sub.InitialConsumerPending()
	if err != nil {
		t.Fatalf("Unexpected error: %v", err)
	}
	if initialPending != 0 {
		t.Fatalf("Expected no initial pending, got %d", initialPending)
	}
	sub.Unsubscribe()

	// Check that Queue subscribe with HB or FC fails.
	_, err = js.QueueSubscribeSync("foo", "wq", nats.IdleHeartbeat(time.Second))