Merge pull request #147 from ploubser/maint_fixes

Remove uses of jsm.go from kv

Author: ploubser

jetstream/resource_kv_bucket.go

@@ -15,12 +15,12 @@ package jetstream
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"time"
 
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/validation"
-	"github.com/nats-io/jsm.go"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 )
@@ -115,7 +115,7 @@ func resourceKVBucket() *schema.Resource {
 }
 
 func resourceKVBucketCreate(d *schema.ResourceData, m any) error {
-	nc, mgr, err := m.(func() (*nats.Conn, *jsm.Manager, error))()
+	nc, err := getConnection(d, m)
 	if err != nil {
 		return err
 	}
@@ -145,14 +145,6 @@ func resourceKVBucketCreate(d *schema.ResourceData, m any) error {
 		}
 	}
 
-	known, err := mgr.IsKnownStream("KV_" + name)
-	if err != nil {
-		return err
-	}
-	if known {
-		return fmt.Errorf("bucket %s already exist", name)
-	}
-
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
 
@@ -161,6 +153,15 @@ func resourceKVBucketCreate(d *schema.ResourceData, m any) error {
 		return err
 	}
 
+	known, err := js.KeyValue(ctx, name)
+	if known != nil {
+		return fmt.Errorf("bucket %s already exist", name)
+	} else if err != nil {
+		if !errors.Is(err, jetstream.ErrBucketNotFound) {
+			return fmt.Errorf("failed to load KV bucket: %s", err)
+		}
+	}
+
 	js.CreateKeyValue(ctx, jetstream.KeyValueConfig{
 		Bucket:         name,
 		Description:    descrption,
@@ -185,7 +186,7 @@ func resourceKVBucketRead(d *schema.ResourceData, m any) error {
 		return err
 	}
 
-	nc, _, err := m.(func() (*nats.Conn, *jsm.Manager, error))()
+	nc, err := getConnection(d, m)
 	if err != nil {
 		return err
 	}
@@ -237,7 +238,7 @@ func resourceKVBucketRead(d *schema.ResourceData, m any) error {
 func resourceKVBucketUpdate(d *schema.ResourceData, m any) error {
 	name := d.Get("name").(string)
 
-	nc, mgr, err := m.(func() (*nats.Conn, *jsm.Manager, error))()
+	nc, err := getConnection(d, m)
 	if err != nil {
 		return err
 	}
@@ -258,29 +259,46 @@ func resourceKVBucketUpdate(d *schema.ResourceData, m any) error {
 	if err != nil {
 		return err
 	}
+
 	jStatus := status.(*jetstream.KeyValueBucketStatus)
 
-	str, err := mgr.LoadStream(jStatus.StreamInfo().Config.Name)
+	str, err := js.Stream(ctx, jStatus.StreamInfo().Config.Name)
 	if err != nil {
 		return err
 	}
 
+	cfg := jetstream.KeyValueConfig{
+		Bucket:         bucket.Bucket(),
+		Description:    str.CachedInfo().Config.Description,
+		MaxValueSize:   str.CachedInfo().Config.MaxMsgSize,
+		History:        uint8(status.History()),
+		TTL:            status.TTL(),
+		MaxBytes:       str.CachedInfo().Config.MaxBytes,
+		Storage:        str.CachedInfo().Config.Storage,
+		Replicas:       str.CachedInfo().Config.Replicas,
+		Placement:      str.CachedInfo().Config.Placement,
+		RePublish:      str.CachedInfo().Config.RePublish,
+		Mirror:         str.CachedInfo().Config.Mirror,
+		Sources:        str.CachedInfo().Config.Sources,
+		Compression:    status.IsCompressed(),
+		LimitMarkerTTL: status.LimitMarkerTTL(),
+	}
+
 	history := d.Get("history").(int)
 	ttl := d.Get("ttl").(int)
 	maxV := d.Get("max_value_size").(int)
 	maxB := d.Get("max_bucket_size").(int)
 	description := d.Get("description").(string)
 	markerTTL := d.Get("limit_marker_ttl").(int)
 
-	cfg := str.Configuration()
-	cfg.MaxAge = time.Duration(ttl) * time.Second
-	cfg.MaxMsgSize = int32(maxV)
+	cfg.History = uint8(history)
+	cfg.TTL = time.Duration(ttl) * time.Second
+	cfg.MaxValueSize = int32(maxV)
 	cfg.MaxBytes = int64(maxB)
-	cfg.MaxMsgsPer = int64(history)
 	cfg.Description = description
-	cfg.SubjectDeleteMarkerTTL = time.Duration(markerTTL) * time.Second
+	cfg.LimitMarkerTTL = time.Duration(markerTTL) * time.Second
 
-	err = str.UpdateConfiguration(cfg)
+	_, err = js.CreateOrUpdateKeyValue(ctx, cfg)
 	if err != nil {
 		return err
 	}
@@ -291,7 +309,7 @@ func resourceKVBucketUpdate(d *schema.ResourceData, m any) error {
 func resourceKVBucketDelete(d *schema.ResourceData, m any) error {
 	name := d.Get("name").(string)
 
-	nc, _, err := m.(func() (*nats.Conn, *jsm.Manager, error))()
+	nc, err := getConnection(d, m)
 	if err != nil {
 		return err
 	}

jetstream/resource_kv_entry.go

@@ -17,7 +17,6 @@ import (
 	"fmt"
 
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
-	"github.com/nats-io/jsm.go"
 	"github.com/nats-io/nats.go"
 )
 
@@ -60,7 +59,7 @@ func resourceKVEntry() *schema.Resource {
 }
 
 func resourceKVEntryCreate(d *schema.ResourceData, m any) error {
-	nc, _, err := m.(func() (*nats.Conn, *jsm.Manager, error))()
+	nc, err := getConnection(d, m)
 	if err != nil {
 		return err
 	}
@@ -94,7 +93,7 @@ func resourceKVEntryRead(d *schema.ResourceData, m any) error {
 		return err
 	}
 
-	nc, _, err := m.(func() (*nats.Conn, *jsm.Manager, error))()
+	nc, err := getConnection(d, m)
 	if err != nil {
 		return err
 	}
@@ -132,7 +131,7 @@ func resourceKVEntryRead(d *schema.ResourceData, m any) error {
 func resourceKVEntryUpdate(d *schema.ResourceData, m any) error {
 	bucket := d.Get("bucket").(string)
 
-	nc, _, err := m.(func() (*nats.Conn, *jsm.Manager, error))()
+	nc, err := getConnection(d, m)
 	if err != nil {
 		return err
 	}
@@ -161,7 +160,7 @@ func resourceKVEntryUpdate(d *schema.ResourceData, m any) error {
 func resourceKVEntryDelete(d *schema.ResourceData, m any) error {
 	bucket := d.Get("bucket").(string)
 
-	nc, _, err := m.(func() (*nats.Conn, *jsm.Manager, error))()
+	nc, err := getConnection(d, m)
 	if err != nil {
 		return err
 	}

jetstream/util.go

@@ -482,6 +482,11 @@ func getConnectProperties(d *schema.ResourceData) (*connectProperties, error) {
 	return &p, nil
 }
 
+func getConnection(d *schema.ResourceData, m any) (*nats.Conn, error) {
+	nc, _, err := m.(func() (*nats.Conn, *jsm.Manager, error))()
+	return nc, err
+}
+
 func connectMgr(d *schema.ResourceData) (any, error) {
 	return func() (*nats.Conn, *jsm.Manager, error) {
 		props, err := getConnectProperties(d)