Function TestJetStreamSyncSubscribeWithMaxAckPending

test/js_test.go:10810–10854

(t *testing.T)

}

func TestJetStreamSyncSubscribeWithMaxAckPending(t *testing.T) {
	opts := natsserver.DefaultTestOptions
	opts.Port = -1
	opts.JetStream = true
	opts.JetStreamLimits.MaxAckPending = 123
	s := RunServerWithOptions(&opts)
	defer shutdownJSServerAndRemoveStorage(t, s)

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

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

	// By default, the sync subscription will be created with a MaxAckPending equal
	// to the internal sync queue len, which is 64K. So that should error out
	// and make sure we get the actual limit

	checkSub := func(pull bool) {
		var sub *nats.Subscription
		var err error
		if pull {
			_, err = js.PullSubscribe("foo", "bar")
		} else {
			_, err = js.SubscribeSync("foo")
		}
		if err == nil || !strings.Contains(err.Error(), "system limit of 123") {
			t.Fatalf("Unexpected error: %v", err)
		}

		// But it should work if we use MaxAckPending() with lower value
		if pull {
			sub, err = js.PullSubscribe("foo", "bar", nats.MaxAckPending(64))
		} else {
			sub, err = js.SubscribeSync("foo", nats.MaxAckPending(64))
		}
		if err != nil {
			t.Fatalf("Unexpected error: %v", err)
		}
		sub.Unsubscribe()
	}
	checkSub(false)
	checkSub(true)
}

func TestJetStreamClusterPlacement(t *testing.T) {
	// There used to be a test here that would not work because it would require