Merge pull request #1553 from ripienaar/durable_and_direct

Use durables for interest and wq streams as well

Author: ripienaar

cli/sub_command.go

@@ -368,8 +368,12 @@ func (c *subCmd) validateInputs(ctx context.Context, nc *nats.Conn, mgr *jsm.Man
 			return err
 		}
 
+		if c.direct && c.durable != "" {
+			return fmt.Errorf("cannot use direct get when a durable name is supplied")
+		}
+
 		config := c.streamObj.CachedInfo().Config
-		c.direct = c.direct || config.Retention == jetstream.WorkQueuePolicy || config.Retention == jetstream.InterestPolicy
+		c.direct = c.durable == "" && (c.direct || config.Retention == jetstream.WorkQueuePolicy || config.Retention == jetstream.InterestPolicy)
 
 		if c.direct {
 			if len(c.subjects) > 1 {

go.mod

@@ -15,13 +15,13 @@ require (
 	github.com/google/go-cmp v0.7.0
 	github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510
 	github.com/gosuri/uiprogress v0.0.1
-	github.com/jedib0t/go-pretty/v6 v6.6.9
+	github.com/jedib0t/go-pretty/v6 v6.7.0
 	github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51
 	github.com/klauspost/compress v1.18.1
 	github.com/nats-io/jsm.go v0.3.1-0.20251103105609-f4ade0fffe41
 	github.com/nats-io/jwt/v2 v2.8.0
-	github.com/nats-io/nats-server/v2 v2.12.1
-	github.com/nats-io/nats.go v1.47.0
+	github.com/nats-io/nats-server/v2 v2.12.1-0.20251107115401-d458bfe6142f
+	github.com/nats-io/nats.go v1.47.1-0.20251107121445-bc5748093780
 	github.com/nats-io/nkeys v0.4.11
 	github.com/nats-io/nuid v1.0.1
 	github.com/prometheus/client_golang v1.23.2
@@ -67,6 +67,7 @@ require (
 	github.com/shopspring/decimal v1.4.0 // indirect
 	github.com/spf13/cast v1.10.0 // indirect
 	github.com/synadia-io/orbit.go/natsext v0.1.1 // indirect
+	go.uber.org/automaxprocs v1.6.0 // indirect
 	go.yaml.in/yaml/v2 v2.4.3 // indirect
 	golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546 // indirect
 	golang.org/x/net v0.46.0 // indirect

go.sum

@@ -72,8 +72,8 @@ github.com/hinshun/vt10x v0.0.0-20220119200601-820417d04eec h1:qv2VnGeEQHchGaZ/u
 github.com/hinshun/vt10x v0.0.0-20220119200601-820417d04eec/go.mod h1:Q48J4R4DvxnHolD5P8pOtXigYlRuPLGl6moFx3ulM68=
 github.com/huandu/xstrings v1.5.0 h1:2ag3IFq9ZDANvthTwTiqSSZLjDc+BedvHPAp5tJy2TI=
 github.com/huandu/xstrings v1.5.0/go.mod h1:y5/lhBue+AyNmUVz9RLU9xbLR0o4KIIExikq4ovT0aE=
-github.com/jedib0t/go-pretty/v6 v6.6.9 h1:PQecJLK3L8ODuVyMe2223b61oRJjrKnmXAncbWTv9MY=
-github.com/jedib0t/go-pretty/v6 v6.6.9/go.mod h1:YwC5CE4fJ1HFUDeivSV1r//AmANFHyqczZk+U6BDALU=
+github.com/jedib0t/go-pretty/v6 v6.7.0 h1:DanoN1RnjXTwDN+B8yqtixXzXqNBCs2Vxo2ARsnrpsY=
+github.com/jedib0t/go-pretty/v6 v6.7.0/go.mod h1:YwC5CE4fJ1HFUDeivSV1r//AmANFHyqczZk+U6BDALU=
 github.com/jung-kurt/gofpdf v1.0.3-0.20190309125859-24315acbbda5/go.mod h1:7Id9E/uU8ce6rXgefFLlgrJj/GYY22cpxn+r32jIOes=
 github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 h1:Z9n2FFNUXsshfwJMBgNA0RU6/i7WVaAegv3PtuIHPMs=
 github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51/go.mod h1:CzGEWj7cYgsdH8dAjBGEr58BoE7ScuLd+fwFZ44+/x8=
@@ -109,20 +109,18 @@ github.com/mitchellh/reflectwalk v1.0.2 h1:G2LzWKi524PWgd3mLHV8Y5k7s6XUvT0Gef6zx
 github.com/mitchellh/reflectwalk v1.0.2/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
-github.com/nats-io/jsm.go v0.3.1-0.20251022125454-e0d9a790a753 h1:IphLag06ffJzMcXjdLUsz3pZ8HDcNOdH22vgkKA7F+s=
-github.com/nats-io/jsm.go v0.3.1-0.20251022125454-e0d9a790a753/go.mod h1:PsAEriiIt8f8/G6qaDt24J913QEOwsds4dKE4MW4rgc=
-github.com/nats-io/jsm.go v0.3.1-0.20251103100619-589746489329 h1:w0vdlBCgW38rcR7dytXuDq5nj+pmgfeGml3no7F/I0A=
-github.com/nats-io/jsm.go v0.3.1-0.20251103100619-589746489329/go.mod h1:vIEz/CZ61EktDu/5XkH46bnN1ku1j+pStDwwEPhy8so=
-github.com/nats-io/jsm.go v0.3.1-0.20251103104243-0be8e60862a2 h1:U8rkI4XHISmdIZgPDWVDIqnW0absA6IL7/gMo8WfLVk=
-github.com/nats-io/jsm.go v0.3.1-0.20251103104243-0be8e60862a2/go.mod h1:vIEz/CZ61EktDu/5XkH46bnN1ku1j+pStDwwEPhy8so=
 github.com/nats-io/jsm.go v0.3.1-0.20251103105609-f4ade0fffe41 h1:S81sizZiW7pauAwQfkdza92THqKudq3n2sk7+WsFyHU=
 github.com/nats-io/jsm.go v0.3.1-0.20251103105609-f4ade0fffe41/go.mod h1:vIEz/CZ61EktDu/5XkH46bnN1ku1j+pStDwwEPhy8so=
 github.com/nats-io/jwt/v2 v2.8.0 h1:K7uzyz50+yGZDO5o772eRE7atlcSEENpL7P+b74JV1g=
 github.com/nats-io/jwt/v2 v2.8.0/go.mod h1:me11pOkwObtcBNR8AiMrUbtVOUGkqYjMQZ6jnSdVUIA=
+github.com/nats-io/nats-server/v2 v2.12.1-0.20251107115401-d458bfe6142f h1:bnAueejTYz9myWfGSNtlfhBgBfwhLTE/2/obZiGeIbk=
+github.com/nats-io/nats-server/v2 v2.12.1-0.20251107115401-d458bfe6142f/go.mod h1:44O2kuwyjKoAsqUL0v5dg4VEi/36Fr52/WlTjnhuWeY=
 github.com/nats-io/nats-server/v2 v2.12.1 h1:0tRrc9bzyXEdBLcHr2XEjDzVpUxWx64aZBm7Rl1QDrA=
 github.com/nats-io/nats-server/v2 v2.12.1/go.mod h1:OEaOLmu/2e6J9LzUt2OuGjgNem4EpYApO5Rpf26HDs8=
 github.com/nats-io/nats.go v1.47.0 h1:YQdADw6J/UfGUd2Oy6tn4Hq6YHxCaJrVKayxxFqYrgM=
 github.com/nats-io/nats.go v1.47.0/go.mod h1:iRWIPokVIFbVijxuMQq4y9ttaBTMe0SFdlZfMDd+33g=
+github.com/nats-io/nats.go v1.47.1-0.20251107121445-bc5748093780 h1:1W5T0ga8fpDklzjM4YrqBhPDUEnmQ+909t1SoVN+Fwk=
+github.com/nats-io/nats.go v1.47.1-0.20251107121445-bc5748093780/go.mod h1:iRWIPokVIFbVijxuMQq4y9ttaBTMe0SFdlZfMDd+33g=
 github.com/nats-io/nkeys v0.4.11 h1:q44qGV008kYd9W1b1nEBkNzvnWxtRSQ7A8BoqRrcfa0=
 github.com/nats-io/nkeys v0.4.11/go.mod h1:szDimtgmfOi9n25JpfIdGw12tZFYXqhGxjhVxsatHVE=
 github.com/nats-io/nsc/v2 v2.12.0 h1:YCs8axEfQkbVLZDuYF4V6aJvPrDmlKJe8mWV+Rgqrzo=

nats/tests/sub_command_test.go

@@ -799,6 +799,19 @@ func TestJetStreamSubscribe(t *testing.T) {
 		})
 	})
 
+	t.Run("--durable and --direct", func(t *testing.T) {
+		withJSServer(t, func(t *testing.T, srv *server.Server, nc *nats.Conn, mgr *jsm.Manager) error {
+			createDefaultTestStream(t, mgr, 1)
+
+			_, err := runNatsCliWithInput(t, "", fmt.Sprintf("--server='%s' sub --stream TEST_STREAM --durable TEST --direct", srv.ClientURL()))
+			if !strings.Contains(err.Error(), "cannot use direct get when a durable name is supplied") {
+				t.Fatalf("expected durable+direct error, got %v", err)
+			}
+
+			return nil
+		})
+	})
+
 	t.Run("--durable with push", func(t *testing.T) {
 		withJSServer(t, func(t *testing.T, srv *server.Server, nc *nats.Conn, mgr *jsm.Manager) error {
 			createDefaultTestStream(t, mgr, 1)
@@ -829,6 +842,34 @@ func TestJetStreamSubscribe(t *testing.T) {
 		})
 	})
 
+	t.Run("--durable with interest streams", func(t *testing.T) {
+		withJSServer(t, func(t *testing.T, srv *server.Server, nc *nats.Conn, mgr *jsm.Manager) error {
+			createDefaultTestStream(t, mgr, 1, jsm.InterestRetention())
+
+			js, err := jetstream.New(nc)
+			checkErr(t, err, "unable to create jetstream context")
+
+			_, err = js.CreateConsumer(context.TODO(), "TEST_STREAM", jetstream.ConsumerConfig{
+				Durable:        "TEST_PUSH",
+				AckPolicy:      jetstream.AckExplicitPolicy,
+				DeliverSubject: nats.NewInbox(),
+				DeliverGroup:   "X",
+			})
+			checkErr(t, err, "unable to create consumer")
+
+			_, err = js.PublishMsg(context.TODO(), defaultTestMsg)
+			checkErr(t, err, "unable to publish message")
+
+			output := string(runNatsCli(t, fmt.Sprintf("--server='%s' sub --stream TEST_STREAM --durable=TEST_PUSH --last --count=1", srv.ClientURL())))
+			if !expectMatchLine(t, output, "Subscribing to JetStream Stream \"TEST_STREAM\" using existing push consumer \"TEST_PUSH\"") ||
+				!expectMatchLine(t, output, primaryTestMsgData) {
+				t.Errorf("unexpected response: %s", output)
+			}
+
+			return nil
+		})
+	})
+
 	t.Run("--headers-only", func(t *testing.T) {
 		withJSServer(t, func(t *testing.T, srv *server.Server, nc *nats.Conn, mgr *jsm.Manager) error {
 			createDefaultTestStream(t, mgr, 1)