Repository: nats-io/nats.go
Pull request #801 (2021-08-18 23:54:51)

Removed new option SubjectIsDelivery and some other changes
This is to align with newer version of the ADR describing the
subscription workflow:
- Removed option SubjectIsDelivery that was introduced in main branch
(but not released yet).
- Take into consideration the new ACK layout
- Make sure that OrderedConsumer rejects user configuring DeliverSubject

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>


Patch:
diff --git a/js.go b/js.go
index e58fb53..cc92f8c 100644
--- a/js.go
+++ b/js.go
@@ -1029,7 +1029,7 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 
 	// If no stream name is specified, or if option SubjectIsDelivery is
 	// specified, the subject cannot be empty.
-	if subj == _EMPTY_ && (o.stream == _EMPTY_ || o.subjIsDelivery) {
+	if subj == _EMPTY_ && o.stream == _EMPTY_ {
 		return nil, fmt.Errorf("nats: subject required")
 	}
 
@@ -1044,7 +1044,7 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 			return nil, fmt.Errorf("nats: invalid ack mode for pull consumers: %s", o.cfg.AckPolicy)
 		}
 		// No deliver subject should be provided
-		if o.cfg.DeliverSubject != _EMPTY_ || o.subjIsDelivery {
+		if o.cfg.DeliverSubject != _EMPTY_ {
 			return nil, ErrPullSubscribeToPushConsumer
 		}
 	}
@@ -1098,6 +1098,10 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 		if o.cfg.MaxDeliver != 1 && o.cfg.MaxDeliver != 0 {
 			return nil, fmt.Errorf("nats: max deliver can not be set for an ordered consumer")
 		}
+		// No deliver subject, we pick our own.
+		if o.cfg.DeliverSubject != _EMPTY_ {
+			return nil, fmt.Errorf("nats: deliver subject can not be set for an ordered consumer")
+		}
 		// Queue groups not allowed.
 		if queue != _EMPTY_ {
 			return nil, fmt.Errorf("nats: queues not be set for an ordered consumer")
@@ -1123,95 +1127,89 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 		hbi = o.cfg.Heartbeat
 	}
 
-	// With this option, we go directly create the NATS subscription
-	// and skip all lookup/create.
-	if o.subjIsDelivery {
-		deliver = subj
-	} else {
-		// In case a consumer has not been set explicitly, then the
-		// durable name will be used as the consumer name.
-		if consumer == _EMPTY_ {
-			consumer = o.cfg.Durable
-		}
+	// In case a consumer has not been set explicitly, then the
+	// durable name will be used as the consumer name.
+	if consumer == _EMPTY_ {
+		consumer = o.cfg.Durable
+	}
 
-		// Find the stream mapped to the subject if not bound to a stream already.
-		if o.stream == _EMPTY_ {
-			stream, err = js.lookupStreamBySubject(subj)
-			if err != nil {
-				return nil, err
-			}
-		} else {
-			stream = o.stream
+	// Find the stream mapped to the subject if not bound to a stream already.
+	if o.stream == _EMPTY_ {
+		stream, err = js.lookupStreamBySubject(subj)
+		if err != nil {
+			return nil, err
 		}
+	} else {
+		stream = o.stream
+	}
 
-		// With an explicit durable name, we can lookup the consumer first
-		// to which it should be attaching to.
-		if consumer != _EMPTY_ {
-			info, err = js.ConsumerInfo(stream, consumer)
-			notFoundErr = errors.Is(err, ErrConsumerNotFound)
-			lookupErr = err == ErrJetStreamNotEnabled || err == ErrTimeout || err == context.DeadlineExceeded
-		}
+	// With an explicit durable name, we can lookup the consumer first
+	// to which it should be attaching to.
+	if consumer != _EMPTY_ {
+		info, err = js.ConsumerInfo(stream, consumer)
+		notFoundErr = errors.Is(err, ErrConsumerNotFound)
+		lookupErr = err == ErrJetStreamNotEnabled || err == ErrTimeout || err == context.DeadlineExceeded
+	}
 
-		switch {
-		case info != nil:
-			deliver, err = processConsInfo(info, isPullMode, subj, queue)
-			if err != nil {
-				return nil, err
-			}
-			icfg := &info.Config
-			hasFC, hbi = icfg.FlowControl, icfg.Heartbeat
-			hasHeartbeats = hbi > 0
-		case (err != nil && !notFoundErr) || (notFoundErr && consumerBound):
-			// If the consumer is being bound and we got an error on pull subscribe then allow the error.
-			if !(isPullMode && lookupErr && consumerBound) {
-				return nil, err
-			}
-		default:
-			// Attempt to create consumer if not found nor using Bind.
-			shouldCreate = true
-			if o.cfg.DeliverSubject != _EMPTY_ {
-				deliver = o.cfg.DeliverSubject
-			} else if !isPullMode {
-				deliver = nc.newInbox()
-				cfg.DeliverSubject = deliver
-			}
+	switch {
+	case info != nil:
+		deliver, err = processConsInfo(info, isPullMode, subj, queue)
+		if err != nil {
+			return nil, err
+		}
+		icfg := &info.Config
+		hasFC, hbi = icfg.FlowControl, icfg.Heartbeat
+		hasHeartbeats = hbi > 0
+	case (err != nil && !notFoundErr) || (notFoundErr && consumerBound):
+		// If the consumer is being bound and we got an error on pull subscribe then allow the error.
+		if !(isPullMode && lookupErr && consumerBound) {
+			return nil, err
+		}
+	default:
+		// Attempt to create consumer if not found nor using Bind.
+		shouldCreate = true
+		if o.cfg.DeliverSubject != _EMPTY_ {
+			deliver = o.cfg.DeliverSubject
+		} else if !isPullMode {
+			deliver = nc.newInbox()
+			cfg.DeliverSubject = deliver
+		}
 
-			// Do filtering always, server will clear as needed.
-			cfg.FilterSubject = subj
+		// Do filtering always, server will clear as needed.
+		cfg.FilterSubject = subj
 
-			// Pass the queue to the consumer config
-			if queue != _EMPTY_ {
-				cfg.DeliverGroup = queue
-			}
+		// Pass the queue to the consumer config
+		if queue != _EMPTY_ {
+			cfg.DeliverGroup = queue
+		}
 
-			// If not set default to ack explicit.
-			if cfg.AckPolicy == ackPolicyNotSet {
-				cfg.AckPolicy = AckExplicitPolicy
-			}
-			// If we have acks at all and the MaxAckPending is not set go ahead
-			// and set to the internal max.
-			// TODO(dlc) - We should be able to update this if client updates PendingLimits.
-			if cfg.MaxAckPending == 0 && cfg.AckPolicy != AckNonePolicy {
-				if !isPullMode && cb != nil && hasFC {
-					cfg.MaxAckPending = DefaultSubPendingMsgsLimit * 16
-				} else if ch != nil {
-					cfg.MaxAckPending = cap(ch)
-				} else {
-					cfg.MaxAckPending = DefaultSubPendingMsgsLimit
-				}
-			}
-			// Create request here.
-			ccreq = &createConsumerRequest{
-				Stream: stream,
-				Config: &cfg,
+		// If not set default to ack explicit.
+		if cfg.AckPolicy == ackPolicyNotSet {
+			cfg.AckPolicy = AckExplicitPolicy
+		}
+		// If we have acks at all and the MaxAckPending is not set go ahead
+		// and set to the internal max.
+		// TODO(dlc) - We should be able to update this if client updates PendingLimits.
+		if cfg.MaxAckPending == 0 && cfg.AckPolicy != AckNonePolicy {
+			if !isPullMode && cb != nil && hasFC {
+				cfg.MaxAckPending = DefaultSubPendingMsgsLimit * 16
+			} else if ch != nil {
+				cfg.MaxAckPending = cap(ch)
+			} else {
+				cfg.MaxAckPending = DefaultSubPendingMsgsLimit
 			}
-			hbi = cfg.Heartbeat
 		}
-
-		if isPullMode {
-			nms = fmt.Sprintf(js.apiSubj(apiRequestNextT), stream, consumer)
-			deliver = nc.newInbox()
+		// Create request here.
+		ccreq = &createConsumerRequest{
+			Stream: stream,
+			Config: &cfg,
 		}
+		hbi = cfg.Heartbeat
+	}
+
+	if isPullMode {
+		nms = fmt.Sprintf(js.apiSubj(apiRequestNextT), stream, consumer)
+		deliver = nc.newInbox()
 	}
 
 	jsi := &jsSub{
@@ -1688,10 +1686,6 @@ type subOpts struct {
 	mack bool
 	// For an ordered consumer.
 	ordered bool
-	// Means that the subject passed to subscribe call will be used
-	// for the low level NATS subscription and no stream nor consumer
-	// lookup/creation will be done.
-	subjIsDelivery bool
 }
 
 // OrderedConsumer will create a fifo direct/ephemeral consumer for in order delivery of messages.
@@ -1919,21 +1913,6 @@ func DeliverSubject(subject string) SubOpt {
 	})
 }
 
-// SubjectIsDelivery specifies that the subject parameter in the subscribe
-// call shall be used to create the NATS subscription and matches the
-// JetStream consumer's deliver subject.
-//
-// NOTE: This is an "expert" API and should only be used when consumer lookup or
-// creation by the library is not possible (for instance cross accounts).
-// Since no lookup of the JetStream consumer is done, there is no way for
-// the library to check the validity of this JetStream subscription.
-func SubjectIsDelivery() SubOpt {
-	return subOptFn(func(opts *subOpts) error {
-		opts.subjIsDelivery = true
-		return nil
-	})
-}
-
 func (sub *Subscription) ConsumerInfo() (*ConsumerInfo, error) {
 	sub.mu.Lock()
 	// TODO(dlc) - Better way to mark especially if we attach.
@@ -2322,12 +2301,12 @@ const (
 )
 
 func getMetadataFields(subject string) ([]string, error) {
-	const noDomainNoHashsExpectedAckTokens = 9
-	const withDomainNoHashExpectedAckTokens = 10
-	const withDomainAndHashExpectedAckTokens = 11
+	const v1TokenCounts = 9
+	const v2TokenCounts = 12
+	const noDomainName = "_"
 
 	const btsep = '.'
-	tsa := [withDomainAndHashExpectedAckTokens]string{}
+	tsa := [v2TokenCounts]string{}
 	start, tokens := 0, tsa[:0]
 	for i := 0; i < len(subject); i++ {
 		if subject[i] == btsep {
@@ -2337,37 +2316,38 @@ func getMetadataFields(subject string) ([]string, error) {
 	}
 	tokens = append(tokens, subject[start:])
 	//
-	// Newer server will include an account hash in the subject, and possibly the domain.
-	// So the subject could be:
+	// Newer server will include the domain name and account hash in the subject,
+	// and a token at the end.
+	//
+	// Old subject was:
+	// $JS.ACK.<stream>.<consumer>.<delivered>.<sseq>.<cseq>.<tm>.<pending>
 	//
-	// no domain: 	$JS.ACK.<account hash>.<stream>.<consumer>...
-	// with domain: $JS.ACK.<domain>.<account hash>.<stream>.<consumer>...
+	// New subject would be:
+	// $JS.ACK.<domain>.<account hash>.<stream>.<consumer>.<delivered>.<sseq>.<cseq>.<tm>.<pending>.<a token with a random value>
 	//
-	// So old server number of tokens is 9, newer is 10 or 11.
+	// v1 has 9 tokens, v2 has 12.
 	//
 	l := len(tokens)
-	if l < noDomainNoHashsExpectedAckTokens || l > withDomainAndHashExpectedAckTokens {
+	// If lower than 9 or more than 9 but less than 12, report an error
+	if l < v1TokenCounts || (l > v1TokenCounts && l < v2TokenCounts) {
 		return nil, ErrNotJSMessage
 	}
 	if tokens[0] != "$JS" || tokens[1] != "ACK" {
 		return nil, ErrNotJSMessage
 	}
-	// To make the rest of the library agnostic of that, we always return the tokens
-	// as if it is coming from a new server will all possible tokens. If domain or account
-	// hash are not specified, the tokens at those locations will simply be empty.
-	if l == noDomainNoHashsExpectedAckTokens || l == withDomainNoHashExpectedAckTokens {
+	// For v1 style, we insert 2 empty tokens (domain and hash) so that the
+	// rest of the library references known fields at a constant location.
+	if l == 9 {
 		// Extend the array (we know the backend is big enough)
-		// Compute how many tokens we need to insert.
-		itc := withDomainAndHashExpectedAckTokens - l
-		for i := 0; i < itc; i++ {
-			tokens = append(tokens, _EMPTY_)
-		}
+		tokens = append(tokens, _EMPTY_, _EMPTY_)
 		// Move to the right anything that is after "ACK" token.
-		copy(tokens[ackDomainTokenPos+itc:], tokens[ackDomainTokenPos:])
-		// Set the missing tokens to empty
-		for i := 0; i < itc; i++ {
-			tokens[ackDomainTokenPos+i] = _EMPTY_
-		}
+		copy(tokens[ackDomainTokenPos+2:], tokens[ackDomainTokenPos:])
+		// Clear the domain and hash tokens
+		tokens[ackDomainTokenPos], tokens[ackAccHashTokenPos] = _EMPTY_, _EMPTY_
+
+	} else if tokens[ackDomainTokenPos] == noDomainName {
+		// If domain is "_", replace with empty value.
+		tokens[ackDomainTokenPos] = _EMPTY_
 	}
 	return tokens, nil
 }


Test patch:
diff --git a/js_test.go b/js_test.go
index b268c56..ed30d09 100644
--- a/js_test.go
+++ b/js_test.go
@@ -133,6 +133,11 @@ func TestJetStreamOrderedConsumer(t *testing.T) {
 		t.Fatalf("Expected an error, got %v", err)
 	}
 
+	_, err = js.SubscribeSync("a", OrderedConsumer(), DeliverSubject("some.subject"))
+	if err == nil || !strings.Contains(err.Error(), "ordered consumer") {
+		t.Fatalf("Expected an error, got %v", err)
+	}
+
 	si, err := js.StreamInfo("OBJECT")
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -570,40 +575,39 @@ func TestJetStreamAckTokens(t *testing.T) {
 		name     string
 		expected *MsgMetadata
 		str      string
+		end      string
 		err      bool
 	}{
 		{
 			"valid token size but not js ack",
 			nil,
-			"one.two.stream.consumer.1.2.3.4.5",
-			true,
-		},
-		{
-			"valid token size but not js ack",
-			nil,
-			"one.two.hash.stream.consumer.1.2.3.4.5",
+			"1.2.3.4.5.6.7.8.9",
+			"",
 			true,
 		},
 		{
 			"valid token size but not js ack",
 			nil,
-			"one.two.domain.hash.stream.consumer.1.2.3.4.5",
+			"1.2.3.4.5.6.7.8.9.10.11.12",
+			"",
 			true,
 		},
 		{
 			"invalid token size",
 			nil,
-			"$JS.ACK.stream.consumer.1.2.3.4",
+			"$JS.ACK.3.4.5.6.7.8",
+			"",
 			true,
 		},
 		{
 			"invalid token size",
 			nil,
-			"$JS.ACK.domain.hash.stream.consumer.1.2.3.4.5.6",
+			"$JS.ACK.3.4.5.6.7.8.9.10.11",
+			"",
 			true,
 		},
 		{
-			"no domain no hash",
+			"v1 style",
 			&MsgMetadata{
 				Stream:       "TEST",
 				Consumer:     "cons",
@@ -616,11 +620,30 @@ func TestJetStreamAckTokens(t *testing.T) {
 				NumPending: 4,
 			},
 			"",
+			"",
+			false,
+		},
+		{
+			"v2 style no domain with hash",
+			&MsgMetadata{
+				Stream:       "TEST",
+				Consumer:     "cons",
+				NumDelivered: 1,
+				Sequence: SequencePair{
+					Stream:   2,
+					Consumer: 3,
+				},
+				Timestamp:  now,
+				NumPending: 4,
+			},
+			"_.ACCHASH.",
+			".abcde",
 			false,
 		},
 		{
-			"no domain with hash",
+			"v2 style with domain and hash",
 			&MsgMetadata{
+				Domain:       "HUB",
 				Stream:       "TEST",
 				Consumer:     "cons",
 				NumDelivered: 1,
@@ -631,11 +654,12 @@ func TestJetStreamAckTokens(t *testing.T) {
 				Timestamp:  now,
 				NumPending: 4,
 			},
-			"ACCHASH.",
+			"HUB.ACCHASH.",
+			".abcde",
 			false,
 		},
 		{
-			"with domain with hash",
+			"more than 12 tokens",
 			&MsgMetadata{
 				Domain:       "HUB",
 				Stream:       "TEST",
@@ -649,6 +673,7 @@ func TestJetStreamAckTokens(t *testing.T) {
 				NumPending: 4,
 			},
 			"HUB.ACCHASH.",
+			".abcde.ghijk.lmnop",
 			false,
 		},
 	} {
@@ -658,7 +683,7 @@ func TestJetStreamAckTokens(t *testing.T) {
 			if test.err {
 				msg.Reply = test.str
 			} else {
-				msg.Reply = fmt.Sprintf("$JS.ACK.%sTEST.cons.1.2.3.%v.4", test.str, now.UnixNano())
+				msg.Reply = fmt.Sprintf("$JS.ACK.%sTEST.cons.1.2.3.%v.4%s", test.str, now.UnixNano(), test.end)
 			}
 
 			meta, err := msg.Metadata()
diff --git a/test/js_test.go b/test/js_test.go
index 4e67f16..864dae8 100644
--- a/test/js_test.go
+++ b/test/js_test.go
@@ -570,11 +570,6 @@ func TestJetStreamSubscribe(t *testing.T) {
 	if err != nats.ErrPullSubscribeToPushConsumer {
 		t.Fatalf("Unexpected error: %v", err)
 	}
-	// Can't specify SubjectIsDelivery() for pull subscribers
-	_, err = js.PullSubscribe("bar", "foo", nats.SubjectIsDelivery())
-	if err != nats.ErrPullSubscribeToPushConsumer {
-		t.Fatalf("Unexpected error: %v", err)
-	}
 	// If stream name is not specified, need the subject.
 	_, err = js.PullSubscribe("", "rip")
 	if err == nil || !strings.Contains(err.Error(), "required") {
@@ -2007,18 +2002,6 @@ func TestJetStreamImportDirectOnly(t *testing.T) {
 	}
 	waitForPending(t, toSend)
 
-	// It is also possible to create a subscription with a SubjectIsDelivery()
-	// option that says that the given subject will be used to create the low
-	// level NATS subscription and no lookup/create attempt will be made.
-	sub, err = js.SubscribeSync("p.d4", nats.SubjectIsDelivery())
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	js.Publish("orders", []byte("msg"))
-	if _, err := sub.NextMsg(time.Second); err != nil {
-		t.Fatalf("Error getting message: %v", err)
-	}
-
 	// Even if there are no permissions or import to check that a consumer exists,
 	// it is still possible to bind subscription to it.
 	sub, err = js.PullSubscribe("ORDERS", "d1", nats.Bind("ORDERS", "d1"))