[IMPROVED] Add extra options to OrderedConsumerConfig

jetstream/consumer_config.go

@@ -263,6 +263,16 @@ type (
 		// Maximum number of attempts for the consumer to be recreated in a
 		// single recreation cycle. Defaults to unlimited.
 		MaxResetAttempts int
+
+		// MaxRequestMaxBytes is the optional maximum total bytes that can be
+		// requested in a given batch. When set with MaxRequestBatch, the batch
+		// size will be constrained by whichever limit is hit first.
+		MaxRequestMaxBytes int `json:"max_bytes,omitempty"`
+
+		// Metadata is a set of application-defined key-value pairs for
+		// associating metadata on the consumer. This feature requires
+		// nats-server v2.10.0 or later.
+		Metadata map[string]string `json:"metadata,omitempty"`
 	}
 
 	// DeliverPolicy determines from which point to start delivering messages.

jetstream/ordered.go

@@ -596,14 +596,16 @@ func (c *orderedConsumer) getConsumerConfig() *ConsumerConfig {
 	}
 	name := fmt.Sprintf("%s_%d", c.namePrefix, c.serial)
 	cfg := &ConsumerConfig{
-		Name:              name,
-		DeliverPolicy:     DeliverByStartSequencePolicy,
-		OptStartSeq:       nextSeq,
-		AckPolicy:         AckNonePolicy,
-		InactiveThreshold: 5 * time.Minute,
-		Replicas:          1,
-		HeadersOnly:       c.cfg.HeadersOnly,
-		MemoryStorage:     true,
+		Name:               name,
+		DeliverPolicy:      DeliverByStartSequencePolicy,
+		OptStartSeq:        nextSeq,
+		AckPolicy:          AckNonePolicy,
+		InactiveThreshold:  5 * time.Minute,
+		Replicas:           1,
+		HeadersOnly:        c.cfg.HeadersOnly,
+		MemoryStorage:      true,
+		MaxRequestMaxBytes: c.cfg.MaxRequestMaxBytes,
+		Metadata:           c.cfg.Metadata,
 	}
 	if len(c.cfg.FilterSubjects) == 1 {
 		cfg.FilterSubject = c.cfg.FilterSubjects[0]

jetstream/test/ordered_test.go

@@ -1946,47 +1946,55 @@ func TestOrderedConsumerConfig(t *testing.T) {
 		{
 			name: "all fields customized, start with custom seq",
 			config: jetstream.OrderedConsumerConfig{
-				FilterSubjects:    []string{"foo.a", "foo.b"},
-				DeliverPolicy:     jetstream.DeliverByStartSequencePolicy,
-				OptStartSeq:       10,
-				ReplayPolicy:      jetstream.ReplayOriginalPolicy,
-				InactiveThreshold: 10 * time.Second,
-				HeadersOnly:       true,
+				FilterSubjects:     []string{"foo.a", "foo.b"},
+				DeliverPolicy:      jetstream.DeliverByStartSequencePolicy,
+				OptStartSeq:        10,
+				ReplayPolicy:       jetstream.ReplayOriginalPolicy,
+				InactiveThreshold:  10 * time.Second,
+				HeadersOnly:        true,
+				MaxRequestMaxBytes: 1024,
+				Metadata:           map[string]string{"foo": "a"},
 			},
 			expected: jetstream.ConsumerConfig{
-				FilterSubjects:    []string{"foo.a", "foo.b"},
-				OptStartSeq:       10,
-				DeliverPolicy:     jetstream.DeliverByStartSequencePolicy,
-				AckPolicy:         jetstream.AckNonePolicy,
-				MaxDeliver:        -1,
-				MaxWaiting:        512,
-				InactiveThreshold: 10 * time.Second,
-				Replicas:          1,
-				MemoryStorage:     true,
-				HeadersOnly:       true,
+				FilterSubjects:     []string{"foo.a", "foo.b"},
+				OptStartSeq:        10,
+				DeliverPolicy:      jetstream.DeliverByStartSequencePolicy,
+				AckPolicy:          jetstream.AckNonePolicy,
+				MaxDeliver:         -1,
+				MaxWaiting:         512,
+				InactiveThreshold:  10 * time.Second,
+				Replicas:           1,
+				MemoryStorage:      true,
+				HeadersOnly:        true,
+				MaxRequestMaxBytes: 1024,
+				Metadata:           map[string]string{"foo": "a"},
 			},
 		},
 		{
 			name: "all fields customized, start with custom time",
 			config: jetstream.OrderedConsumerConfig{
-				FilterSubjects:    []string{"foo.a", "foo.b"},
-				DeliverPolicy:     jetstream.DeliverByStartTimePolicy,
-				OptStartTime:      &time.Time{},
-				ReplayPolicy:      jetstream.ReplayOriginalPolicy,
-				InactiveThreshold: 10 * time.Second,
-				HeadersOnly:       true,
+				FilterSubjects:     []string{"foo.a", "foo.b"},
+				DeliverPolicy:      jetstream.DeliverByStartTimePolicy,
+				OptStartTime:       &time.Time{},
+				ReplayPolicy:       jetstream.ReplayOriginalPolicy,
+				InactiveThreshold:  10 * time.Second,
+				HeadersOnly:        true,
+				MaxRequestMaxBytes: 1024,
+				Metadata:           map[string]string{"foo": "a"},
 			},
 			expected: jetstream.ConsumerConfig{
-				FilterSubjects:    []string{"foo.a", "foo.b"},
-				OptStartTime:      &time.Time{},
-				DeliverPolicy:     jetstream.DeliverByStartTimePolicy,
-				AckPolicy:         jetstream.AckNonePolicy,
-				MaxDeliver:        -1,
-				MaxWaiting:        512,
-				InactiveThreshold: 10 * time.Second,
-				Replicas:          1,
-				MemoryStorage:     true,
-				HeadersOnly:       true,
+				FilterSubjects:     []string{"foo.a", "foo.b"},
+				OptStartTime:       &time.Time{},
+				DeliverPolicy:      jetstream.DeliverByStartTimePolicy,
+				AckPolicy:          jetstream.AckNonePolicy,
+				MaxDeliver:         -1,
+				MaxWaiting:         512,
+				InactiveThreshold:  10 * time.Second,
+				Replicas:           1,
+				MemoryStorage:      true,
+				HeadersOnly:        true,
+				MaxRequestMaxBytes: 1024,
+				Metadata:           map[string]string{"foo": "a"},
 			},
 		},
 	}