Merge pull request #66 from nats-io/stream_repub

support stream republish

Author: ripienaar

README.md

@@ -196,6 +196,9 @@ resource "jetstream_consumer" "ORDERS_NEW" {
  * `replicas` - (optional) How many replicas of the data to keep in a clustered environment
  * `memory` - (optional) Force the consumer state to be kept in memory rather than inherit the setting from the stream
  * `backoff` - (optional) List of durations in Go format that represents a retry time scale for NaK'd messages. A list of durations in seconds
+ * `republish_source` - (optional) Republish matching messages to `republish_destination`
+ * `republish_destination` - (optional) The destination to publish messages to
+ * `republish_headers_only` - (optional) Republish only message headers, no bodies
 
 ## jetstream_kv_bucket
 

docs/resources/jetstream_stream.md

@@ -60,3 +60,6 @@ Above the `ORDERS_ARCHIVE` stream is a mirror of `ORDERS`, valid options for spe
  * `deny_purge` - (optional) Restricts the ability to purge messages from a stream via the API. Cannot be change once set to true (bool)
  * `allow_rollup_hdrs` - (optional) Allows the use of the Nats-Rollup header to replace all contents of a stream, or subject in a stream, with a single new message (bool)
  * `allow_direct` - (optional) Allow higher performance, direct access to get individual messages via the $JS.DS.GET API (bool)
+ * `republish_source` - (optional) Republish matching messages to `republish_destination`
+ * `republish_destination` - (optional) The destination to publish messages to
+ * `republish_headers_only` - (optional) Republish only message headers, no bodies

jetstream/resource_jetstream_stream.go

@@ -213,6 +213,25 @@ func resourceStream() *schema.Resource {
 				Optional:    true,
 				Elem:        &schema.Resource{Schema: sourceInfo},
 			},
+			"republish_source": {
+				Type:        schema.TypeString,
+				Description: "Republish messages to republish_destination",
+				ForceNew:    false,
+				Optional:    true,
+			},
+			"republish_destination": {
+				Type:         schema.TypeString,
+				Description:  "The destination to publish messages to",
+				ForceNew:     false,
+				Optional:     true,
+				RequiredWith: []string{"republish_source"},
+			},
+			"republish_headers_only": {
+				Type:        schema.TypeBool,
+				Description: "Republish only message headers, no bodies",
+				ForceNew:    false,
+				Optional:    true,
+			},
 		},
 	}
 }
@@ -340,6 +359,13 @@ func resourceStreamRead(d *schema.ResourceData, m any) error {
 			}
 		}
 	}
+
+	if str.IsRepublishing() {
+		d.Set("republish_source", str.Republish().Source)
+		d.Set("republish_destination", str.Republish().Destination)
+		d.Set("republish_headers_only", str.Republish().HeadersOnly)
+	}
+
 	return nil
 }
 

jetstream/util.go

@@ -162,6 +162,17 @@ func streamConfigFromResourceData(d *schema.ResourceData) (cfg api.StreamConfig,
 		Placement:     placement,
 	}
 
+	repubSrc := d.Get("republish_source").(string)
+	repubDest := d.Get("republish_destination").(string)
+	repubHdrs := d.Get("republish_headers_only").(bool)
+	if repubSrc != "" {
+		stream.RePublish = &api.RePublish{
+			Source:      repubSrc,
+			Destination: repubDest,
+			HeadersOnly: repubHdrs,
+		}
+	}
+
 	if description, ok := d.GetOk("description"); ok {
 		stream.Description = description.(string)
 	}