Feature implementation from commits 92def3b..7bfd96a

README.md

@@ -23,7 +23,7 @@ A [Go](http://golang.org) client for the [NATS messaging system](https://nats.io
 go get github.com/nats-io/nats.go@latest
 
 # To get a specific version:
-go get github.com/nats-io/nats.go@v1.40.1
+go get github.com/nats-io/nats.go@v1.41.1
 
 # Note that the latest major version for NATS Server is v2:
 go get github.com/nats-io/nats-server/v2@latest

go.mod

@@ -6,7 +6,6 @@ require (
 	github.com/klauspost/compress v1.18.0
 	github.com/nats-io/nkeys v0.4.9
 	github.com/nats-io/nuid v1.0.1
-	golang.org/x/text v0.23.0
 )
 
 require (

go.sum

@@ -8,5 +8,3 @@ golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U=
 golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
 golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
 golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/text v0.23.0 h1:D71I7dUrlY+VX0gQShAThNGHFxZ13dGLBHQLVl1mJlY=
-golang.org/x/text v0.23.0/go.mod h1:/BLNzu4aZCJ1+kcD0DNRotWKage4q2rGVAg4o22unh4=

jetstream/api.go

@@ -1,4 +1,4 @@
-// Copyright 2022-2023 The NATS Authors
+// Copyright 2022-2025 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -104,6 +104,9 @@ const (
 
 	// apiMsgDeleteT is the endpoint to remove a message.
 	apiMsgDeleteT = "STREAM.MSG.DELETE.%s"
+
+	// apiConsumerUnpinT is the endpoint to unpin a consumer.
+	apiConsumerUnpinT = "CONSUMER.UNPIN.%s.%s"
 )
 
 func (js *jetStream) apiRequestJSON(ctx context.Context, subject string, resp any, data ...[]byte) (*jetStreamMsg, error) {

jetstream/consumer.go

@@ -1,4 +1,4 @@
-// Copyright 2022-2024 The NATS Authors
+// Copyright 2022-2025 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -374,3 +374,37 @@ func validateConsumerName(dur string) error {
 	}
 	return nil
 }
+
+func unpinConsumer(ctx context.Context, js *jetStream, stream, consumer, group string) error {
+	ctx, cancel := js.wrapContextWithoutDeadline(ctx)
+	if cancel != nil {
+		defer cancel()
+	}
+	if err := validateConsumerName(consumer); err != nil {
+		return err
+	}
+	unpinSubject := fmt.Sprintf(apiConsumerUnpinT, stream, consumer)
+
+	var req = consumerUnpinRequest{
+		Group: group,
+	}
+
+	reqJSON, err := json.Marshal(req)
+	if err != nil {
+		return err
+	}
+
+	var resp apiResponse
+
+	if _, err := js.apiRequestJSON(ctx, unpinSubject, &resp, reqJSON); err != nil {
+		return err
+	}
+	if resp.Error != nil {
+		if resp.Error.ErrorCode == JSErrCodeConsumerNotFound {
+			return ErrConsumerNotFound
+		}
+		return resp.Error
+	}
+
+	return nil
+}

jetstream/consumer_config.go

@@ -76,6 +76,9 @@ type (
 		// TimeStamp indicates when the info was gathered by the server.
 		TimeStamp time.Time `json:"ts"`
 
+		// PriorityGroups contains the information about the currently defined priority groups
+		PriorityGroups []PriorityGroupState `json:"priority_groups,omitempty"`
+
 		// Paused indicates whether the consumer is paused.
 		Paused bool `json:"paused,omitempty"`
 
@@ -84,6 +87,17 @@ type (
 		PauseRemaining time.Duration `json:"pause_remaining,omitempty"`
 	}
 
+	PriorityGroupState struct {
+		// Group this status is for.
+		Group string `json:"group"`
+
+		// PinnedClientID is the generated ID of the pinned client.
+		PinnedClientID string `json:"pinned_client_id,omitempty"`
+
+		// PinnedTS is the timestamp when the client was pinned.
+		PinnedTS time.Time `json:"pinned_ts,omitempty"`
+	}
+
 	// ConsumerConfig is the configuration of a JetStream consumer.
 	ConsumerConfig struct {
 		// Name is an optional name for the consumer. If not set, one is
@@ -227,6 +241,18 @@ type (
 
 		// PauseUntil is for suspending the consumer until the deadline.
 		PauseUntil *time.Time `json:"pause_until,omitempty"`
+
+		// PriorityPolicy represents he priority policy the consumer is set to.
+		// Requires nats-server v2.11.0 or later.
+		PriorityPolicy PriorityPolicy `json:"priority_policy,omitempty"`
+
+		// PinnedTTL represents the time after which the client will be unpinned
+		// if no new pull requests are sent.Used with PriorityPolicyPinned.
+		// Requires nats-server v2.11.0 or later.
+		PinnedTTL time.Duration `json:"priority_timeout,omitempty"`
+
+		// PriorityGroups is a list of priority groups this consumer supports.
+		PriorityGroups []string `json:"priority_groups,omitempty"`
 	}
 
 	// OrderedConsumerConfig is the configuration of an ordered JetStream
@@ -263,7 +289,7 @@ type (
 
 		// InactiveThreshold is a duration which instructs the server to clean
 		// up the consumer if it has been inactive for the specified duration.
-		// Defaults to 5s.
+		// Defaults to 5m.
 		InactiveThreshold time.Duration `json:"inactive_threshold,omitempty"`
 
 		// HeadersOnly indicates whether only headers of messages should be sent
@@ -298,8 +324,51 @@ type (
 		Stream   uint64     `json:"stream_seq"`
 		Last     *time.Time `json:"last_active,omitempty"`
 	}
+
+	// PriorityPolicy determines the priority policy the consumer is set to.
+	PriorityPolicy int
 )
 
+const (
+	// PriorityPolicyNone is the default priority policy.
+	PriorityPolicyNone PriorityPolicy = iota
+
+	// PriorityPolicyPinned is the priority policy that pins a consumer to a
+	// specific client.
+	PriorityPolicyPinned
+
+	// PriorityPolicyOverflow is the priority policy that allows for
+	// restricting when a consumer will receive messages based on the number of
+	// pending messages or acks.
+	PriorityPolicyOverflow
+)
+
+func (p *PriorityPolicy) UnmarshalJSON(data []byte) error {
+	switch string(data) {
+	case jsonString(""):
+		*p = PriorityPolicyNone
+	case jsonString("pinned_client"):
+		*p = PriorityPolicyPinned
+	case jsonString("overflow"):
+		*p = PriorityPolicyOverflow
+	default:
+		return fmt.Errorf("nats: can not unmarshal %q", data)
+	}
+	return nil
+}
+
+func (p PriorityPolicy) MarshalJSON() ([]byte, error) {
+	switch p {
+	case PriorityPolicyNone:
+		return json.Marshal("")
+	case PriorityPolicyPinned:
+		return json.Marshal("pinned_client")
+	case PriorityPolicyOverflow:
+		return json.Marshal("overflow")
+	}
+	return nil, fmt.Errorf("nats: unknown priority policy %v", p)
+}
+
 const (
 	// DeliverAllPolicy starts delivering messages from the very beginning of a
 	// stream. This is the default.

jetstream/errors.go

@@ -195,6 +195,10 @@ var (
 	// consumer.
 	ErrNoMessages JetStreamError = &jsError{message: "no messages"}
 
+	// ErrPinIDMismatch is returned when Pin ID sent in the request does not match
+	// the currently pinned consumer subscriber ID on the server.
+	ErrPinIDMismatch JetStreamError = &jsError{message: "pin ID mismatch"}
+
 	// ErrMaxBytesExceeded is returned when a message would exceed MaxBytes set
 	// on a pull request.
 	ErrMaxBytesExceeded JetStreamError = &jsError{message: "message size exceeds max bytes"}

jetstream/jetstream.go

@@ -304,6 +304,11 @@ type (
 		// Domain is the domain name token used when sending JetStream requests.
 		Domain string
 
+		// DefaultTimeout is the default timeout used for JetStream API requests.
+		// This applies when the context passed to JetStream methods does not have
+		// a deadline set.
+		DefaultTimeout time.Duration
+
 		publisherOpts asyncPublisherOpts
 
 		// this is the actual prefix used in the API requests
@@ -407,6 +412,7 @@ func New(nc *nats.Conn, opts ...JetStreamOpt) (JetStream, error) {
 		publisherOpts: asyncPublisherOpts{
 			maxpa: defaultAsyncPubAckInflight,
 		},
+		DefaultTimeout: defaultAPITimeout,
 	}
 	setReplyPrefix(nc, &jsOpts)
 	for _, opt := range opts {
@@ -451,7 +457,8 @@ func NewWithAPIPrefix(nc *nats.Conn, apiPrefix string, opts ...JetStreamOpt) (Je
 		publisherOpts: asyncPublisherOpts{
 			maxpa: defaultAsyncPubAckInflight,
 		},
-		APIPrefix: apiPrefix,
+		APIPrefix:      apiPrefix,
+		DefaultTimeout: defaultAPITimeout,
 	}
 	setReplyPrefix(nc, &jsOpts)
 	for _, opt := range opts {
@@ -488,7 +495,8 @@ func NewWithDomain(nc *nats.Conn, domain string, opts ...JetStreamOpt) (JetStrea
 		publisherOpts: asyncPublisherOpts{
 			maxpa: defaultAsyncPubAckInflight,
 		},
-		Domain: domain,
+		Domain:         domain,
+		DefaultTimeout: defaultAPITimeout,
 	}
 	setReplyPrefix(nc, &jsOpts)
 	for _, opt := range opts {
@@ -1089,7 +1097,7 @@ func (js *jetStream) wrapContextWithoutDeadline(ctx context.Context) (context.Co
 	if _, ok := ctx.Deadline(); ok {
 		return ctx, nil
 	}
-	return context.WithTimeout(ctx, defaultAPITimeout)
+	return context.WithTimeout(ctx, js.opts.DefaultTimeout)
 }
 
 // CleanupPublisher will cleanup the publishing side of JetStreamContext.

jetstream/jetstream_options.go

@@ -65,6 +65,19 @@ func WithPublishAsyncTimeout(dur time.Duration) JetStreamOpt {
 	}
 }
 
+// WithDefaultTimeout sets the default timeout for JetStream API requests.
+// It is used when context used for the request does not have a deadline set.
+// If not provided, a default of 5 seconds will be used.
+func WithDefaultTimeout(timeout time.Duration) JetStreamOpt {
+	return func(opts *JetStreamOptions) error {
+		if timeout <= 0 {
+			return fmt.Errorf("%w: timeout value must be greater than 0", ErrInvalidOption)
+		}
+		opts.DefaultTimeout = timeout
+		return nil
+	}
+}
+
 // WithPurgeSubject sets a specific subject for which messages on a stream will
 // be purged
 func WithPurgeSubject(subject string) StreamPurgeOpt {
@@ -284,6 +297,73 @@ func (t PullThresholdBytes) configureMessages(opts *consumeOpts) error {
 	return nil
 }
 
+// PullMinPending sets the minimum number of messages that should be pending for
+// a consumer with PriorityPolicyOverflow to be considered for delivery.
+// If provided, PullPriorityGroup must be set as well and the consumer has to have
+// PriorityPolicy set to PriorityPolicyOverflow.
+//
+// PullMinPending implements both PullConsumeOpt and PullMessagesOpt, allowing
+// it to configure Consumer.Consume and Consumer.Messages.
+type PullMinPending int
+
+func (min PullMinPending) configureConsume(opts *consumeOpts) error {
+	if min < 1 {
+		return fmt.Errorf("%w: min pending should be more than 0", ErrInvalidOption)
+	}
+	opts.MinPending = int64(min)
+	return nil
+}
+
+func (min PullMinPending) configureMessages(opts *consumeOpts) error {
+	if min < 1 {
+		return fmt.Errorf("%w: min pending should be more than 0", ErrInvalidOption)
+	}
+	opts.MinPending = int64(min)
+	return nil
+}
+
+// PullMinAckPending sets the minimum number of pending acks that should be
+// present for a consumer with PriorityPolicyOverflow to be considered for
+// delivery. If provided, PullPriorityGroup must be set as well and the consumer
+// has to have PriorityPolicy set to PriorityPolicyOverflow.
+//
+// PullMinAckPending implements both PullConsumeOpt and PullMessagesOpt, allowing
+// it to configure Consumer.Consume and Consumer.Messages.
+type PullMinAckPending int
+
+func (min PullMinAckPending) configureConsume(opts *consumeOpts) error {
+	if min < 1 {
+		return fmt.Errorf("%w: min pending should be more than 0", ErrInvalidOption)
+	}
+	opts.MinAckPending = int64(min)
+	return nil
+}
+
+func (min PullMinAckPending) configureMessages(opts *consumeOpts) error {
+	if min < 1 {
+		return fmt.Errorf("%w: min pending should be more than 0", ErrInvalidOption)
+	}
+	opts.MinAckPending = int64(min)
+	return nil
+}
+
+// PullPriorityGroup sets the priority group for a consumer.
+// It has to match one of the priority groups set on the consumer.
+//
+// PullPriorityGroup implements both PullConsumeOpt and PullMessagesOpt, allowing
+// it to configure Consumer.Consume and Consumer.Messages.
+type PullPriorityGroup string
+
+func (g PullPriorityGroup) configureConsume(opts *consumeOpts) error {
+	opts.Group = string(g)
+	return nil
+}
+
+func (g PullPriorityGroup) configureMessages(opts *consumeOpts) error {
+	opts.Group = string(g)
+	return nil
+}
+
 // PullHeartbeat sets the idle heartbeat duration for a pull subscription
 // If a client does not receive a heartbeat message from a stream for more
 // than the idle heartbeat setting, the subscription will be removed
@@ -355,6 +435,43 @@ func WithMessagesErrOnMissingHeartbeat(hbErr bool) PullMessagesOpt {
 	})
 }
 
+// FetchMinPending sets the minimum number of messages that should be pending for
+// a consumer with PriorityPolicyOverflow to be considered for delivery.
+// If provided, FetchPriorityGroup must be set as well and the consumer has to have
+// PriorityPolicy set to PriorityPolicyOverflow.
+func FetchMinPending(min int64) FetchOpt {
+	return func(req *pullRequest) error {
+		if min < 1 {
+			return fmt.Errorf("%w: min pending should be more than 0", ErrInvalidOption)
+		}
+		req.MinPending = min
+		return nil
+	}
+}
+
+// FetchMinAckPending sets the minimum number of pending acks that should be
+// present for a consumer with PriorityPolicyOverflow to be considered for
+// delivery. If provided, FetchPriorityGroup must be set as well and the consumer
+// has to have PriorityPolicy set to PriorityPolicyOverflow.
+func FetchMinAckPending(min int64) FetchOpt {
+	return func(req *pullRequest) error {
+		if min < 1 {
+			return fmt.Errorf("%w: min ack pending should be more than 0", ErrInvalidOption)
+		}
+		req.MinAckPending = min
+		return nil
+	}
+}
+
+// FetchPriorityGroup sets the priority group for a consumer.
+// It has to match one of the priority groups set on the consumer.
+func FetchPriorityGroup(group string) FetchOpt {
+	return func(req *pullRequest) error {
+		req.Group = group
+		return nil
+	}
+}
+
 // FetchMaxWait sets custom timeout for fetching predefined batch of messages.
 //
 // If not provided, a default of 30 seconds will be used.