Merge pull request #794 from nats-io/js_changes

JetStream: lot of changes

.travis.yml

@@ -15,5 +15,5 @@ before_script:
 - find . -type f -name "*.go" | xargs misspell -error -locale US
 - staticcheck ./...
 script:
-- go test -modfile=go_test.mod -v -run=TestNoRace -p=1 ./... --failfast
-- if [[ "$TRAVIS_GO_VERSION" =~ 1.16 ]]; then ./scripts/cov.sh TRAVIS; else go test -modfile=go_test.mod -race -v -p=1 ./... --failfast; fi
+- go test -modfile=go_test.mod -v -run=TestNoRace -p=1 ./... --failfast -vet=off
+- if [[ "$TRAVIS_GO_VERSION" =~ 1.16 ]]; then ./scripts/cov.sh TRAVIS; else go test -modfile=go_test.mod -race -v -p=1 ./... --failfast -vet=off; fi

context.go

@@ -122,12 +122,7 @@ func (s *Subscription) nextMsgWithContext(ctx context.Context, pullSubInternal,
 	}
 
 	s.mu.Lock()
-	err := s.validateNextMsgState()
-	// Unless this is from an internal call, reject use of this API.
-	// Users should use Fetch() instead.
-	if err == nil && !pullSubInternal && s.jsi != nil && s.jsi.pull {
-		err = ErrTypeSubscription
-	}
+	err := s.validateNextMsgState(pullSubInternal)
 	if err != nil {
 		s.mu.Unlock()
 		return nil, err

example_test.go

@@ -324,6 +324,16 @@ func ExampleJetStream() {
 
 	// Async queue subscription where members load balance the
 	// received messages together.
+	// If no consumer name is specified, either with nats.Bind()
+	// or nats.Durable() options, the queue name is used as the
+	// durable name (that is, as if you were passing the
+	// nats.Durable(<queue group name>) option.
+	// It is recommended to use nats.Bind() or nats.Durable()
+	// and preferably create the JetStream consumer beforehand
+	// (using js.AddConsumer) so that the JS consumer is not
+	// deleted on an Unsubscribe() or Drain() when the member
+	// that created the consumer goes away first.
+	// Check Godoc for the QueueSubscribe() API for more details.
 	js.QueueSubscribe("foo", "group", func(msg *nats.Msg) {
 		msg.Ack()
 	}, nats.ManualAck())
@@ -333,7 +343,8 @@ func ExampleJetStream() {
 	msg, _ := sub.NextMsg(2 * time.Second)
 	msg.Ack()
 
-	// QueueSubscribe with group or load balancing.
+	// We can add a member to the group, with this member using
+	// the synchronous version of the QueueSubscribe.
 	sub, _ = js.QueueSubscribeSync("foo", "group")
 	msg, _ = sub.NextMsg(2 * time.Second)
 	msg.Ack()

go_test.mod

@@ -4,7 +4,7 @@ go 1.16
 
 require (
 	github.com/golang/protobuf v1.4.2
-	github.com/nats-io/nats-server/v2 v2.3.4
+	github.com/nats-io/nats-server/v2 v2.3.5-0.20210815013007-eb8aeb217178
 	github.com/nats-io/nkeys v0.3.0
 	github.com/nats-io/nuid v1.0.1
 	google.golang.org/protobuf v1.23.0

go_test.sum

@@ -17,9 +17,9 @@ github.com/nats-io/jwt v1.2.2 h1:w3GMTO969dFg+UOKTmmyuu7IGdusK+7Ytlt//OYH/uU=
 github.com/nats-io/jwt v1.2.2/go.mod h1:/xX356yQA6LuXI9xWW7mZNpxgF2mBmGecH+Fj34sP5Q=
 github.com/nats-io/jwt/v2 v2.0.3 h1:i/O6cmIsjpcQyWDYNcq2JyZ3/VTF8SJ4JWluI5OhpvI=
 github.com/nats-io/jwt/v2 v2.0.3/go.mod h1:VRP+deawSXyhNjXmxPCHskrR6Mq50BqpEI5SEcNiGlY=
-github.com/nats-io/nats-server/v2 v2.3.4 h1:WcNa6HDFX8gjZPHb8CJ9wxRHEjJSlhWUb/MKb6/mlUY=
-github.com/nats-io/nats-server/v2 v2.3.4/go.mod h1:3mtbaN5GkCo/Z5T3nNj0I0/W1fPkKzLiDC6jjWJKp98=
-github.com/nats-io/nats.go v1.11.1-0.20210623165838-4b75fc59ae30/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
+github.com/nats-io/nats-server/v2 v2.3.5-0.20210815013007-eb8aeb217178 h1:6/bt9zMGA1D/i3ROeq8GjF8Tig5BVFh4V3gI+qpoWIs=
+github.com/nats-io/nats-server/v2 v2.3.5-0.20210815013007-eb8aeb217178/go.mod h1:7mTh0KSxKc63xAVop97cFCIGRkWCv6HoX9lMXRSNOhU=
+github.com/nats-io/nats.go v1.11.1-0.20210813172934-9c00d13a8a19/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
 github.com/nats-io/nkeys v0.2.0/go.mod h1:XdZpAbhgyyODYqjTawOnIOI7VlbKSarI9Gfy1tqEu/s=
 github.com/nats-io/nkeys v0.3.0 h1:cgM5tL53EvYRU+2YLXIK0G2mJtK12Ft9oeooSZMA2G8=
 github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV6A5y4=