Add support for consumers with multiple filters

Signed-off-by: Tomasz Pietrek <[email protected]>

controllers/jetstream/consumer.go

@@ -308,7 +308,6 @@ func consumerSpecToOpts(spec apis.ConsumerSpec) ([]jsm.ConsumerOption, error) {
 	opts := []jsm.ConsumerOption{
 		jsm.DurableName(spec.DurableName),
 		jsm.DeliverySubject(spec.DeliverSubject),
-		jsm.FilterStreamBySubject(spec.FilterSubject),
 		jsm.RateLimitBitsPerSecond(uint64(spec.RateLimitBps)),
 		jsm.MaxAckPending(uint(spec.MaxAckPending)),
 		jsm.ConsumerDescription(spec.Description),
@@ -319,6 +318,16 @@ func consumerSpecToOpts(spec apis.ConsumerSpec) ([]jsm.ConsumerOption, error) {
 		jsm.ConsumerOverrideReplicas(spec.Replicas),
 	}
 
+	if spec.FilterSubject != "" && len(spec.FilterSubjects) > 0 {
+		return nil, fmt.Errorf("cannot specify both FilterSubject and FilterSubjects")
+	}
+
+	if spec.FilterSubject != "" {
+		opts = append(opts, jsm.FilterStreamBySubject(spec.FilterSubject))
+	} else if len(spec.FilterSubjects) > 0 {
+		opts = append(opts, jsm.FilterStreamBySubject(spec.FilterSubjects...))
+	}
+
 	switch spec.DeliverPolicy {
 	case "all":
 		opts = append(opts, jsm.DeliverAllAvailable())

deploy/crds.yml

@@ -501,6 +501,11 @@ spec:
               filterSubject:
                 description: Select only a specific incoming subjects, supports wildcards.
                 type: string
+              filterSubjects:
+                description: List of incoming subjects, supports wildcards. Available since 2.10.
+                type: array
+                items:
+                  type: string
               replayPolicy:
                 description: How messages are sent.
                 type: string

pkg/jetstream/apis/jetstream/v1beta2/consumertypes.go

@@ -34,6 +34,7 @@ type ConsumerSpec struct {
 	PreventUpdate      bool     `json:"preventUpdate"`
 	DurableName        string   `json:"durableName"`
 	FilterSubject      string   `json:"filterSubject"`
+	FilterSubjects     []string `json:"filterSubjects"`
 	FlowControl        bool     `json:"flowControl"`
 	HeadersOnly        bool     `json:"headersOnly"`
 	HeartbeatInterval  string   `json:"heartbeatInterval"`