Repository: nats-io/nats.go
Pull request #1339 (2023-07-03 14:28:52)

Fix typos and doc comments for JetStream simplified API
Just fixed some typos in the doc comments and updated doc comments for `Consume()` and `Messages()` with the accepted functional options.


Patch:
diff --git a/jetstream/errors.go b/jetstream/errors.go
index 7125c12..ebfdebc 100644
--- a/jetstream/errors.go
+++ b/jetstream/errors.go
@@ -62,7 +62,7 @@ const (
 var (
 	// API errors
 
-	// ErrJetStreamNotEnabled is an error returned when JetStream is not enabled for an account.
+	// ErrJetStreamNotEnabled is an error returned when JetStream is not enabled.
 	ErrJetStreamNotEnabled JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeJetStreamNotEnabled, Description: "jetstream not enabled", Code: 503}}
 
 	// ErrJetStreamNotEnabledForAccount is an error returned when JetStream is not enabled for an account.
@@ -71,13 +71,13 @@ var (
 	// ErrStreamNotFound is an error returned when stream with given name does not exist.
 	ErrStreamNotFound JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeStreamNotFound, Description: "stream not found", Code: 404}}
 
-	// ErrStreamNameAlreadyInUse is returned when a stream with given name already exists and has a different configuration
+	// ErrStreamNameAlreadyInUse is returned when a stream with given name already exists and has a different configuration.
 	ErrStreamNameAlreadyInUse JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeStreamNameInUse, Description: "stream name already in use", Code: 400}}
 
 	// ErrConsumerNotFound is an error returned when consumer with given name does not exist.
 	ErrConsumerNotFound JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeConsumerNotFound, Description: "consumer not found", Code: 404}}
 
-	// ErrMsgNotFound is returned when message with provided sequence number does npt exist.
+	// ErrMsgNotFound is returned when message with provided sequence number does not exist.
 	ErrMsgNotFound JetStreamError = &jsError{apiErr: &APIError{ErrorCode: JSErrCodeMessageNotFound, Description: "message not found", Code: 404}}
 
 	// ErrBadRequest is returned when invalid request is sent to JetStream API.
@@ -167,7 +167,7 @@ var (
 	// messages using Fetch (or FetchBytes).
 	ErrOrderConsumerUsedAsFetch = &jsError{message: "ordered consumer initialized as fetch"}
 
-	// ErrOrderConsumerUsedAsFetch is returned when ordered consumer was already used to process
+	// ErrOrderConsumerUsedAsConsume is returned when ordered consumer was already used to process
 	// messages using Consume or Messages.
 	ErrOrderConsumerUsedAsConsume = &jsError{message: "ordered consumer initialized as consume"}
 
diff --git a/jetstream/jetstream.go b/jetstream/jetstream.go
index 36fad23..51a2858 100644
--- a/jetstream/jetstream.go
+++ b/jetstream/jetstream.go
@@ -50,10 +50,10 @@ type (
 		// PublishMsg performs a synchronous publish to a stream and waits for ack from server
 		// It accepts subject name (which must be bound to a stream) and nats.Message
 		PublishMsg(ctx context.Context, msg *nats.Msg, opts ...PublishOpt) (*PubAck, error)
-		// PublishAsync performs a asynchronous publish to a stream and returns [PubAckFuture] interface
+		// PublishAsync performs an asynchronous publish to a stream and returns [PubAckFuture] interface
 		// It accepts subject name (which must be bound to a stream) and message data
 		PublishAsync(subject string, payload []byte, opts ...PublishOpt) (PubAckFuture, error)
-		// PublishMsgAsync performs a asynchronous publish to a stream and returns [PubAckFuture] interface
+		// PublishMsgAsync performs an asynchronous publish to a stream and returns [PubAckFuture] interface
 		// It accepts subject name (which must be bound to a stream) and nats.Message
 		PublishMsgAsync(msg *nats.Msg, opts ...PublishOpt) (PubAckFuture, error)
 		// PublishAsyncPending returns the number of async publishes outstanding for this context
@@ -197,11 +197,11 @@ type (
 
 var subjectRegexp = regexp.MustCompile(`^[^ >]*[>]?$`)
 
-// New returns a enw JetStream instance
+// New returns a new JetStream instance.
 //
 // Available options:
-// [WithClientTrace] - enables request/response tracing
-// [WithPublishAsyncErrHandler] - sets error handler for async message publish
+// [WithClientTrace] - enables request/response tracing.
+// [WithPublishAsyncErrHandler] - sets error handler for async message publish.
 // [WithPublishAsyncMaxPending] - sets the maximum outstanding async publishes that can be inflight at one time.
 // [WithDirectGet] - specifies whether client should use direct get requests.
 func New(nc *nats.Conn, opts ...JetStreamOpt) (JetStream, error) {
diff --git a/jetstream/message.go b/jetstream/message.go
index 36216c5..3d6bf5e 100644
--- a/jetstream/message.go
+++ b/jetstream/message.go
@@ -166,12 +166,12 @@ func (m *jetStreamMsg) Headers() nats.Header {
 	return m.msg.Header
 }
 
-// Subject reutrns a subject on which a message is published
+// Subject returns a subject on which a message is published
 func (m *jetStreamMsg) Subject() string {
 	return m.msg.Subject
 }
 
-// Reply reutrns a reply subject for a JetStream message
+// Reply returns a reply subject for a JetStream message
 func (m *jetStreamMsg) Reply() string {
 	return m.msg.Reply
 }
diff --git a/jetstream/options.go b/jetstream/options.go
index 7b8f1fe..3c97425 100644
--- a/jetstream/options.go
+++ b/jetstream/options.go
@@ -56,7 +56,7 @@ func WithPublishAsyncMaxPending(max int) JetStreamOpt {
 	}
 }
 
-// WithPurgeSubject sets a sprecific subject for which messages on a stream will be purged
+// WithPurgeSubject sets a specific subject for which messages on a stream will be purged
 func WithPurgeSubject(subject string) StreamPurgeOpt {
 	return func(req *StreamPurgeRequest) error {
 		req.Subject = subject
@@ -64,7 +64,7 @@ func WithPurgeSubject(subject string) StreamPurgeOpt {
 	}
 }
 
-// WithPurgeSequence is used to set a sprecific sequence number up to which (but not including) messages will be purged from a stream
+// WithPurgeSequence is used to set a specific sequence number up to which (but not including) messages will be purged from a stream
 // Can be combined with [WithPurgeSubject] option, but not with [WithPurgeKeep]
 func WithPurgeSequence(sequence uint64) StreamPurgeOpt {
 	return func(req *StreamPurgeRequest) error {
@@ -206,8 +206,7 @@ func ConsumeErrHandler(cb ConsumeErrHandlerFunc) PullConsumeOpt {
 	})
 }
 
-// ConsumeErrHandler sets custom error handler invoked when an error was encountered while consuming messages
-// It will be invoked for both terminal (Consumer Deleted, invalid request body) and non-terminal (e.g. missing heartbeats) errors
+// WithMessagesErrOnMissingHeartbeat sets whether a missing heartbeat error should be reported when calling Next (Default: true).
 func WithMessagesErrOnMissingHeartbeat(hbErr bool) PullMessagesOpt {
 	return pullOptFunc(func(cfg *consumeOpts) error {
 		cfg.ReportMissingHeartbeats = hbErr
@@ -215,7 +214,7 @@ func WithMessagesErrOnMissingHeartbeat(hbErr bool) PullMessagesOpt {
 	})
 }
 
-// FetchMaxWait sets custom timeout fir fetching predefined batch of messages
+// FetchMaxWait sets custom timeout for fetching predefined batch of messages
 func FetchMaxWait(timeout time.Duration) FetchOpt {
 	return func(req *pullRequest) error {
 		if timeout <= 0 {
@@ -274,7 +273,7 @@ func WithExpectLastSequencePerSubject(seq uint64) PublishOpt {
 	}
 }
 
-// ExpectLastMsgId sets the expected last msgId in the response from the publish.
+// WithExpectLastMsgID sets the expected last msgId in the response from the publish.
 func WithExpectLastMsgID(id string) PublishOpt {
 	return func(opts *pubOpts) error {
 		opts.lastMsgID = id
diff --git a/jetstream/publish.go b/jetstream/publish.go
index 1b49d2f..e80f638 100644
--- a/jetstream/publish.go
+++ b/jetstream/publish.go
@@ -110,7 +110,7 @@ type (
 )
 
 const (
-	// Default time wait between retries on Publish iff err is NoResponders.
+	// Default time wait between retries on Publish if err is NoResponders.
 	DefaultPubRetryWait = 250 * time.Millisecond
 
 	// Default number of retries
diff --git a/jetstream/pull.go b/jetstream/pull.go
index 28ff24a..abb9cef 100644
--- a/jetstream/pull.go
+++ b/jetstream/pull.go
@@ -30,7 +30,7 @@ import (
 type (
 	// MessagesContext supports iterating over a messages on a stream.
 	MessagesContext interface {
-		// Next retreives nest message on a stream. It will block until the next message is available.
+		// Next retreives next message on a stream. It will block until the next message is available.
 		Next() (Msg, error)
 		// Stop closes the iterator and cancels subscription.
 		Stop()
@@ -136,13 +136,13 @@ const (
 // Consume returns a ConsumeContext, allowing for processing incoming messages from a stream in a given callback function.
 //
 // Available options:
-// [ConsumeMaxMessages] - sets maximum number of messages stored in a buffer, default is set to 100
-// [ConsumeMaxBytes] - sets maximum number of bytes stored in a buffer
-// [ConsumeExpiry] - sets a timeout for individual batch request, default is set to 30 seconds
-// [ConsumeHeartbeat] - sets an idle heartbeat setting for a pull request, default is set to 5s
+// [PullMaxMessages] - sets maximum number of messages stored in a buffer, default is set to 100
+// [PullMaxBytes] - sets maximum number of bytes stored in a buffer
+// [PullExpiry] - sets a timeout for individual batch request, default is set to 30 seconds
+// [PullHeartbeat] - sets an idle heartbeat setting for a pull request, default is set to 5s
 // [ConsumeErrHandler] - sets custom consume error callback handler
-// [ConsumeThresholdMessages] - sets the byte count on which Consume will trigger new pull request to the server
-// [ConsumeThresholdBytes] - sets the message count on which Consume will trigger new pull request to the server
+// [PullThresholdMessages] - sets the message count on which Consume will trigger new pull request to the server
+// [PullThresholdBytes] - sets the byte count on which Consume will trigger new pull request to the server
 func (p *pullConsumer) Consume(handler MessageHandler, opts ...PullConsumeOpt) (ConsumeContext, error) {
 	if handler == nil {
 		return nil, ErrHandlerRequired
@@ -350,13 +350,11 @@ func (s *pullSubscription) checkPending() {
 // Messages returns MessagesContext, allowing continuously iterating over messages on a stream.
 //
 // Available options:
-// [ConsumeMaxMessages] - sets maximum number of messages stored in a buffer, default is set to 100
-// [ConsumeMaxBytes] - sets maximum number of bytes stored in a buffer
-// [ConsumeExpiry] - sets a timeout for individual batch request, default is set to 30 seconds
-// [ConsumeHeartbeat] - sets an idle heartbeat setting for a pull request, default is set to 5s
-// [ConsumeErrHandler] - sets custom consume error callback handler
-// [ConsumeThresholdMessages] - sets the byte count on which Consume will trigger new pull request to the server
-// [ConsumeThresholdBytes] - sets the message count on which Consume will trigger new pull request to the server
+// [PullMaxMessages] - sets maximum number of messages stored in a buffer, default is set to 100
+// [PullMaxBytes] - sets maximum number of bytes stored in a buffer
+// [PullExpiry] - sets a timeout for individual batch request, default is set to 30 seconds
+// [PullHeartbeat] - sets an idle heartbeat setting for a pull request, default is set to 5s
+// [WithMessagesErrOnMissingHeartbeat] - sets whether a missing heartbeat error should be reported when calling Next
 func (p *pullConsumer) Messages(opts ...PullMessagesOpt) (MessagesContext, error) {
 	consumeOpts, err := parseMessagesOpts(opts...)
 	if err != nil {
@@ -596,7 +594,7 @@ func (p *pullConsumer) FetchBytes(maxBytes int, opts ...FetchOpt) (MessageBatch,
 	return p.fetch(req)
 }
 
-// Fetch sends a single request to retrieve given number of messages.
+// FetchNoWait sends a single request to retrieve given number of messages.
 // If there are any messages available at the time of sending request,
 // FetchNoWait will return immediately.
 func (p *pullConsumer) FetchNoWait(batch int) (MessageBatch, error) {
diff --git a/jetstream/stream.go b/jetstream/stream.go
index 8e4ba63..4f29d3b 100644
--- a/jetstream/stream.go
+++ b/jetstream/stream.go
@@ -71,7 +71,7 @@ type (
 		// ListConsumers returns ConsumerInfoLister enabling iterating over a channel of consumer infos
 		ListConsumers(context.Context) ConsumerInfoLister
 
-		// ConsumerNames returns a  ConsumerNameLister enabling iterating over a channel of consumer names
+		// ConsumerNames returns a ConsumerNameLister enabling iterating over a channel of consumer names
 		ConsumerNames(context.Context) ConsumerNameLister
 	}
 	RawStreamMsg struct {
@@ -276,8 +276,8 @@ func (s *stream) CachedInfo() *StreamInfo {
 // Purge removes messages from a stream
 //
 // Available options:
-// [WithPurgeSubject] - can be used set a sprecific subject for which messages on a stream will be purged
-// [WithPurgeSequence] - can be used to set a sprecific sequence number up to which (but not including) messages will be purged from a stream
+// [WithPurgeSubject] - can be used set a specific subject for which messages on a stream will be purged
+// [WithPurgeSequence] - can be used to set a specific sequence number up to which (but not including) messages will be purged from a stream
 // [WithPurgeKeep] - can be used to set the number of messages to be kept in the stream after purge.
 func (s *stream) Purge(ctx context.Context, opts ...StreamPurgeOpt) error {
 	var purgeReq StreamPurgeRequest
@@ -506,7 +506,7 @@ func (s *consumerLister) Err() <-chan error {
 	return s.errs
 }
 
-// ConsumerNames returns a  ConsumerNameLister enabling iterating over a channel of consumer names
+// ConsumerNames returns a ConsumerNameLister enabling iterating over a channel of consumer names
 func (s *stream) ConsumerNames(ctx context.Context) ConsumerNameLister {
 	l := &consumerLister{
 		js:    s.jetStream,
diff --git a/jetstream/stream_config.go b/jetstream/stream_config.go
index 901523c..9b5461d 100644
--- a/jetstream/stream_config.go
+++ b/jetstream/stream_config.go
@@ -110,7 +110,7 @@ type (
 	}
 
 	// RePublish is for republishing messages once committed to a stream. The original
-	// subject cis remapped from the subject pattern to the destination pattern.
+	// subject is remapped from the subject pattern to the destination pattern.
 	RePublish struct {
 		Source      string `json:"src,omitempty"`
 		Destination string `json:"dest"`
@@ -165,7 +165,7 @@ const (
 	// DiscardOld will remove older messages to return to the limits. This is
 	// the default.
 	DiscardOld DiscardPolicy = iota
-	//DiscardNew will fail to store new messages.
+	// DiscardNew will fail to store new messages.
 	DiscardNew
 )
 
diff --git a/js.go b/js.go
index 9ecf1f5..3559f5a 100644
--- a/js.go
+++ b/js.go
@@ -1991,6 +1991,9 @@ func (sub *Subscription) resetOrderedConsumer(sseq uint64) {
 		cfg.DeliverSubject = newDeliver
 		cfg.DeliverPolicy = DeliverByStartSequencePolicy
 		cfg.OptStartSeq = sseq
+		// In case the consumer was created with a start time, we need to clear it
+		// since we are now using a start sequence.
+		cfg.OptStartTime = nil
 
 		js := jsi.js
 		sub.mu.Unlock()


Test patch:
diff --git a/js_test.go b/js_test.go
index c73a261..f01547e 100644
--- a/js_test.go
+++ b/js_test.go
@@ -114,6 +114,9 @@ func TestJetStreamOrderedConsumer(t *testing.T) {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 
+	// Will be used as start time to validate proper reset to sequence on retries.
+	startTime := time.Now()
+
 	// Create a sample asset.
 	msg := make([]byte, 1024*1024)
 	rand.Read(msg)
@@ -186,7 +189,7 @@ func TestJetStreamOrderedConsumer(t *testing.T) {
 			rmsg = append(rmsg, m.Data...)
 		}
 		// OrderedConsumer does not need HB, it sets it on its own, but for test we override which is ok.
-		sub, err := js.Subscribe("a", cb, OrderedConsumer(), IdleHeartbeat(250*time.Millisecond))
+		sub, err := js.Subscribe("a", cb, OrderedConsumer(), IdleHeartbeat(250*time.Millisecond), StartTime(startTime))
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
@@ -208,7 +211,7 @@ func TestJetStreamOrderedConsumer(t *testing.T) {
 		var rmsg []byte
 
 		// OrderedConsumer does not need HB, it sets it on its own, but for test we override which is ok.
-		sub, err := js.SubscribeSync("a", OrderedConsumer(), IdleHeartbeat(250*time.Millisecond))
+		sub, err := js.SubscribeSync("a", OrderedConsumer(), IdleHeartbeat(250*time.Millisecond), StartTime(startTime))
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}