Merge pull request #104 from iapain/main

[FEAT]: adds FilterSubjects

Author: ripienaar

README.md

@@ -187,6 +187,7 @@ resource "jetstream_consumer" "ORDERS_NEW" {
  * `delivery_group` - (optional) When set Push consumers will only deliver messages to subscriptions with this group set
  * `durable_name` - The durable name of the Consumer
  * `filter_subject` - (optional) Only receive a subset of messages from the Stream based on the subject they entered the Stream on
+ * `filter_subjects` - (optional) Only receive a subset tof messages from the Stream based on subjects they entered the Stream on. This is exclusive to `filter_subject`. Only works with v2.10 or better.
  * `max_delivery` - (optional) Maximum deliveries to attempt for each message
  * `replay_policy` - (optional) The rate at which messages will be replayed from the stream
  * `sample_freq` - (optional) The percentage of acknowledgements that will be sampled for observability purposes

docs/resources/jetstream_consumer.md

@@ -33,6 +33,7 @@ resource "jetstream_consumer" "ORDERS_NEW" {
  * `delivery_group` - (optional) When set Push consumers will only deliver messages to subscriptions with this group set
  * `durable_name` - The durable name of the Consumer
  * `filter_subject` - (optional) Only receive a subset of messages from the Stream based on the subject they entered the Stream on
+ * `filter_subjects` - (optional) Only receive a subset tof messages from the Stream based on subjects they entered the Stream on. This is exclusive to `filter_subject`. Only works with v2.10 or better.
  * `max_delivery` - (optional) Maximum deliveries to attempt for each message
  * `replay_policy` - (optional) The rate at which messages will be replayed from the stream
  * `sample_freq` - (optional) The percentage of acknowledgements that will be sampled for observability purposes

jetstream/resource_jetstream_consumer.go

@@ -131,11 +131,19 @@ func resourceConsumer() *schema.Resource {
 				ForceNew:    false,
 			},
 			"filter_subject": {
-				Type:        schema.TypeString,
-				Description: "Only receive a subset of messages from the Stream based on the subject they entered the Stream on",
-				Default:     "",
+				Type:          schema.TypeString,
+				Description:   "Only receive a subset of messages from the Stream based on the subject they entered the Stream on",
+				Default:       "",
+				Optional:      true,
+				ForceNew:      false,
+				ConflictsWith: []string{"filter_subjects"},
+			},
+			"filter_subjects": {
+				Type:        schema.TypeList,
+				Description: "Only receive a subset of messages from the stream baseed on the subjects they entered the Streeam on, exlusive to filter_subject and works with nats-server v2.10 or better",
 				Optional:    true,
 				ForceNew:    false,
+				Elem:        &schema.Schema{Type: schema.TypeString},
 			},
 			"replay_policy": {
 				Type:         schema.TypeString,
@@ -327,6 +335,22 @@ func consumerConfigFromResourceData(d *schema.ResourceData) (cfg api.ConsumerCon
 		cfg.AckPolicy = api.AckNone
 	}
 
+	fs, ok := d.GetOk("filter_subjects")
+	if ok {
+		ns := fs.([]any)
+		if len(ns) == 1 {
+			cfg.FilterSubject = ns[0].(string)
+			cfg.FilterSubjects = nil
+		} else if len(ns) > 1 {
+			var subjects = make([]string, len(ns))
+			for i, v := range ns {
+				subjects[i] = v.(string)
+			}
+			cfg.FilterSubjects = subjects
+			cfg.FilterSubject = ""
+		}
+	}
+
 	m, ok := d.GetOk("metadata")
 	if ok {
 		mt, ok := m.(map[string]any)
@@ -398,6 +422,8 @@ func resourceConsumerUpdate(d *schema.ResourceData, m any) error {
 	opts := []jsm.ConsumerOption{
 		jsm.ConsumerDescription(cfg.Description),
 		jsm.ConsumerMetadata(cfg.Metadata),
+		jsm.FilterStreamBySubject(cfg.FilterSubject),
+		jsm.FilterStreamBySubject(cfg.FilterSubjects...),
 		jsm.AckWait(cfg.AckWait),
 		jsm.MaxDeliveryAttempts(cfg.MaxDeliver),
 		jsm.SamplePercent(freq),
@@ -513,6 +539,7 @@ func resourceConsumerRead(d *schema.ResourceData, m any) error {
 	d.Set("ack_wait", cons.AckWait().Seconds())
 	d.Set("max_delivery", cons.MaxDeliver())
 	d.Set("filter_subject", cons.FilterSubject())
+	d.Set("filter_subjects", cons.FilterSubjects())
 	d.Set("stream_sequence", 0)
 	d.Set("start_time", "")
 	d.Set("ratelimit", cons.RateLimit())

jetstream/resource_jetstream_consumer_test.go

@@ -49,6 +49,26 @@ resource "jetstream_consumer" "TEST_C2" {
   durable_name    = "C2"
   stream_sequence = 10
   max_ack_pending = 20
+  filter_subjects = ["TEST.a", "TEST.b"]
+}
+`
+
+const testConsumerConfig_singleSubject = `
+provider "jetstream" {
+  servers = "%s"
+}
+
+resource "jetstream_stream" "test" {
+  name     = "TEST"
+  subjects = ["TEST.*"]
+}
+
+resource "jetstream_consumer" "TEST_C3" {
+  stream_id       = jetstream_stream.test.id
+  durable_name    = "C3"
+  stream_sequence = 10
+  max_ack_pending = 20
+  filter_subject = "TEST.a"
 }
 `
 
@@ -101,11 +121,21 @@ func TestResourceConsumer(t *testing.T) {
 				Check: resource.ComposeTestCheckFunc(
 					testStreamExist(t, mgr, "TEST"),
 					testConsumerExist(t, mgr, "TEST", "C2"),
+					testConsumerHasFilterSubjects(t, mgr, "TEST", "C2", []string{"TEST.a", "TEST.b"}),
 					resource.TestCheckResourceAttr("jetstream_consumer.TEST_C2", "stream_sequence", "10"),
 					resource.TestCheckResourceAttr("jetstream_consumer.TEST_C2", "max_ack_pending", "20"),
 					resource.TestCheckResourceAttr("jetstream_consumer.TEST_C2", "inactive_threshold", "0"),
 				),
 			},
+			{
+				Config: fmt.Sprintf(testConsumerConfig_singleSubject, nc.ConnectedUrl()),
+				Check: resource.ComposeTestCheckFunc(
+					testStreamExist(t, mgr, "TEST"),
+					testConsumerExist(t, mgr, "TEST", "C3"),
+					testConsumerHasFilterSubjects(t, mgr, "TEST", "C3", []string{}),
+					resource.TestCheckResourceAttr("jetstream_consumer.TEST_C3", "filter_subject", "TEST.a"),
+				),
+			},
 		},
 	})
 }

jetstream/util_test.go

@@ -41,6 +41,23 @@ func testConsumerHasMetadata(t *testing.T, mgr *jsm.Manager, stream string, cons
 	}
 }
 
+func testConsumerHasFilterSubjects(t *testing.T, mgr *jsm.Manager, stream string, consumer string, subjects []string) resource.TestCheckFunc {
+	return func(s *terraform.State) error {
+		cons, err := mgr.LoadConsumer(stream, consumer)
+		if err != nil {
+			return err
+		}
+		if len(cons.FilterSubjects()) == 0 && len(subjects) == 0 {
+			return nil
+		}
+		if cmp.Equal(cons.FilterSubjects(), subjects) {
+			return nil
+		}
+
+		return fmt.Errorf("expected %q got %q", subjects, cons.FilterSubjects())
+	}
+}
+
 func testStreamHasSubjects(t *testing.T, mgr *jsm.Manager, stream string, subjects []string) resource.TestCheckFunc {
 	return func(s *terraform.State) error {
 		str, err := mgr.LoadStream(stream)