Repository: nats-io/nats.go
Pull request #1099 (2022-10-05 12:50:14)

[CHANGED] Listing Key Value stores and object stores to return status instead of interface
This PR addresses https://github.com/nats-io/nats-architecture-and-design/issues/154

This PR changes the signatures of `KeyValueStores()` and `ObjectStores()` methods. For KV, this is not a breaking change since this functionality is yet to be released. For object store, **this is a breaking change** since these methods were released as part of v1.17.0.


Patch:
diff --git a/js.go b/js.go
index 012cb79..05555cc 100644
--- a/js.go
+++ b/js.go
@@ -67,9 +67,11 @@ type JetStream interface {
 	// gone.
 	// * If Durable() option is specified, the library will attempt to lookup a JetStream
 	// consumer with this name, and if found, will bind to it and not attempt to
-	// delete it. However, if not found, the library will send a request to create
-	// such durable JetStream consumer. The library will delete the JetStream consumer
-	// after an Unsubscribe() or Drain().
+	// delete it. However, if not found, the library will send a request to
+	// create such durable JetStream consumer. Note that the library will delete
+	// the JetStream consumer after an Unsubscribe() or Drain() only if it
+	// created the durable consumer while subscribing. If the durable consumer
+	// already existed prior to subscribing it won't be deleted.
 	// * If Bind() option is provided, the library will attempt to lookup the
 	// consumer with the given name, and if successful, bind to it. If the lookup fails,
 	// then the Subscribe() call will return an error.
diff --git a/jsm.go b/jsm.go
index 0bdbca2..d568f25 100644
--- a/jsm.go
+++ b/jsm.go
@@ -99,29 +99,30 @@ type JetStreamManager interface {
 // There are sensible defaults for most. If no subjects are
 // given the name will be used as the only subject.
 type StreamConfig struct {
-	Name              string          `json:"name"`
-	Description       string          `json:"description,omitempty"`
-	Subjects          []string        `json:"subjects,omitempty"`
-	Retention         RetentionPolicy `json:"retention"`
-	MaxConsumers      int             `json:"max_consumers"`
-	MaxMsgs           int64           `json:"max_msgs"`
-	MaxBytes          int64           `json:"max_bytes"`
-	Discard           DiscardPolicy   `json:"discard"`
-	MaxAge            time.Duration   `json:"max_age"`
-	MaxMsgsPerSubject int64           `json:"max_msgs_per_subject"`
-	MaxMsgSize        int32           `json:"max_msg_size,omitempty"`
-	Storage           StorageType     `json:"storage"`
-	Replicas          int             `json:"num_replicas"`
-	NoAck             bool            `json:"no_ack,omitempty"`
-	Template          string          `json:"template_owner,omitempty"`
-	Duplicates        time.Duration   `json:"duplicate_window,omitempty"`
-	Placement         *Placement      `json:"placement,omitempty"`
-	Mirror            *StreamSource   `json:"mirror,omitempty"`
-	Sources           []*StreamSource `json:"sources,omitempty"`
-	Sealed            bool            `json:"sealed,omitempty"`
-	DenyDelete        bool            `json:"deny_delete,omitempty"`
-	DenyPurge         bool            `json:"deny_purge,omitempty"`
-	AllowRollup       bool            `json:"allow_rollup_hdrs,omitempty"`
+	Name                 string          `json:"name"`
+	Description          string          `json:"description,omitempty"`
+	Subjects             []string        `json:"subjects,omitempty"`
+	Retention            RetentionPolicy `json:"retention"`
+	MaxConsumers         int             `json:"max_consumers"`
+	MaxMsgs              int64           `json:"max_msgs"`
+	MaxBytes             int64           `json:"max_bytes"`
+	Discard              DiscardPolicy   `json:"discard"`
+	DiscardNewPerSubject bool            `json:"discard_new_per_subject,omitempty"`
+	MaxAge               time.Duration   `json:"max_age"`
+	MaxMsgsPerSubject    int64           `json:"max_msgs_per_subject"`
+	MaxMsgSize           int32           `json:"max_msg_size,omitempty"`
+	Storage              StorageType     `json:"storage"`
+	Replicas             int             `json:"num_replicas"`
+	NoAck                bool            `json:"no_ack,omitempty"`
+	Template             string          `json:"template_owner,omitempty"`
+	Duplicates           time.Duration   `json:"duplicate_window,omitempty"`
+	Placement            *Placement      `json:"placement,omitempty"`
+	Mirror               *StreamSource   `json:"mirror,omitempty"`
+	Sources              []*StreamSource `json:"sources,omitempty"`
+	Sealed               bool            `json:"sealed,omitempty"`
+	DenyDelete           bool            `json:"deny_delete,omitempty"`
+	DenyPurge            bool            `json:"deny_purge,omitempty"`
+	AllowRollup          bool            `json:"allow_rollup_hdrs,omitempty"`
 
 	// Allow republish of the message after being sequenced and stored.
 	RePublish *RePublish `json:"republish,omitempty"`
@@ -186,7 +187,7 @@ type AccountInfo struct {
 	Tier
 	Domain string          `json:"domain"`
 	API    APIStats        `json:"api"`
-	Tiers  map[string]Tier `json:"tier"`
+	Tiers  map[string]Tier `json:"tiers"`
 }
 
 type Tier struct {
diff --git a/kv.go b/kv.go
index 735b323..4993df9 100644
--- a/kv.go
+++ b/kv.go
@@ -37,8 +37,8 @@ type KeyValueManager interface {
 	DeleteKeyValue(bucket string) error
 	// KeyValueStoreNames is used to retrieve a list of key value store names
 	KeyValueStoreNames() <-chan string
-	// KeyValueStores is used to retrieve a list of key value stores
-	KeyValueStores() <-chan KeyValue
+	// KeyValueStores is used to retrieve a list of key value store statuses
+	KeyValueStores() <-chan KeyValueStatus
 }
 
 // Notice: Experimental Preview
@@ -992,9 +992,9 @@ func (js *js) KeyValueStoreNames() <-chan string {
 	return ch
 }
 
-// KeyValueStores is used to retrieve a list of key value stores
-func (js *js) KeyValueStores() <-chan KeyValue {
-	ch := make(chan KeyValue)
+// KeyValueStores is used to retrieve a list of key value store statuses
+func (js *js) KeyValueStores() <-chan KeyValueStatus {
+	ch := make(chan KeyValueStatus)
 	l := &streamLister{js: js}
 	l.js.opts.streamListSubject = fmt.Sprintf(kvSubjectsTmpl, "*")
 	go func() {
@@ -1004,7 +1004,7 @@ func (js *js) KeyValueStores() <-chan KeyValue {
 				if !strings.HasPrefix(info.Config.Name, "KV_") {
 					continue
 				}
-				ch <- mapStreamToKVS(js, info)
+				ch <- &KeyValueBucketStatus{nfo: info, bucket: strings.TrimPrefix(info.Config.Name, "KV_")}
 			}
 		}
 	}()
diff --git a/object.go b/object.go
index 0cbb0bc..133001f 100644
--- a/object.go
+++ b/object.go
@@ -46,8 +46,8 @@ type ObjectStoreManager interface {
 	DeleteObjectStore(bucket string) error
 	// ObjectStoreNames is used to retrieve a list of bucket names
 	ObjectStoreNames(opts ...ObjectOpt) <-chan string
-	// ObjectStores is used to retrieve a list of buckets
-	ObjectStores(opts ...ObjectOpt) <-chan ObjectStore
+	// ObjectStores is used to retrieve a list of bucket statuses
+	ObjectStores(opts ...ObjectOpt) <-chan ObjectStoreStatus
 }
 
 // ObjectStore is a blob store capable of storing large objects efficiently in
@@ -1314,8 +1314,8 @@ func (js *js) ObjectStoreNames(opts ...ObjectOpt) <-chan string {
 	return ch
 }
 
-// ObjectStores is used to retrieve a list of buckets
-func (js *js) ObjectStores(opts ...ObjectOpt) <-chan ObjectStore {
+// ObjectStores is used to retrieve a list of bucket statuses
+func (js *js) ObjectStores(opts ...ObjectOpt) <-chan ObjectStoreStatus {
 	var o objOpts
 	for _, opt := range opts {
 		if opt != nil {
@@ -1324,7 +1324,7 @@ func (js *js) ObjectStores(opts ...ObjectOpt) <-chan ObjectStore {
 			}
 		}
 	}
-	ch := make(chan ObjectStore)
+	ch := make(chan ObjectStoreStatus)
 	var cancel context.CancelFunc
 	if o.ctx == nil {
 		o.ctx, cancel = context.WithTimeout(context.Background(), defaultRequestWait)
@@ -1343,7 +1343,10 @@ func (js *js) ObjectStores(opts ...ObjectOpt) <-chan ObjectStore {
 					continue
 				}
 				select {
-				case ch <- &obs{name: strings.TrimPrefix(info.Config.Name, "OBJ_"), stream: info.Config.Name, js: js}:
+				case ch <- &ObjectBucketStatus{
+					nfo:    info,
+					bucket: strings.TrimPrefix(info.Config.Name, "OBJ_"),
+				}:
 				case <-o.ctx.Done():
 					return
 				}


Test patch:
diff --git a/go_test.mod b/go_test.mod
index bc0d685..1d94d2b 100644
--- a/go_test.mod
+++ b/go_test.mod
@@ -4,17 +4,17 @@ go 1.17
 
 require (
 	github.com/golang/protobuf v1.4.2
-	github.com/nats-io/nats-server/v2 v2.9.0
+	github.com/nats-io/nats-server/v2 v2.9.2
 	github.com/nats-io/nkeys v0.3.0
 	github.com/nats-io/nuid v1.0.1
 	google.golang.org/protobuf v1.23.0
 )
 
 require (
-	github.com/klauspost/compress v1.15.9 // indirect
+	github.com/klauspost/compress v1.15.11 // indirect
 	github.com/minio/highwayhash v1.0.2 // indirect
 	github.com/nats-io/jwt/v2 v2.3.0 // indirect
-	golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90 // indirect
-	golang.org/x/sys v0.0.0-20220906135438-9e1f76180b77 // indirect
-	golang.org/x/time v0.0.0-20220722155302-e5dcc9cfc0b9 // indirect
+	golang.org/x/crypto v0.0.0-20220926161630-eccd6366d1be // indirect
+	golang.org/x/sys v0.0.0-20220928140112-f11e5e49a4ec // indirect
+	golang.org/x/time v0.0.0-20220922220347-f3bd1da661af // indirect
 )
diff --git a/go_test.sum b/go_test.sum
index 6c704c8..dab6e1c 100644
--- a/go_test.sum
+++ b/go_test.sum
@@ -11,8 +11,8 @@ github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMyw
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.4.0 h1:xsAVV57WRhGj6kEIi8ReJzQlHHqcBYCElAvkovg3B/4=
 github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-github.com/klauspost/compress v1.15.9 h1:wKRjX6JRtDdrE9qwa4b/Cip7ACOshUI4smpCQanqjSY=
-github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU=
+github.com/klauspost/compress v1.15.11 h1:Lcadnb3RKGin4FYM/orgq0qde+nc15E5Cbqg4B9Sx9c=
+github.com/klauspost/compress v1.15.11/go.mod h1:QPwzmACJjUTFsnSHH934V6woptycfrDDJnH7hvFVbGM=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
@@ -20,9 +20,9 @@ github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA
 github.com/minio/highwayhash v1.0.2/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
 github.com/nats-io/jwt/v2 v2.3.0 h1:z2mA1a7tIf5ShggOFlR1oBPgd6hGqcDYsISxZByUzdI=
 github.com/nats-io/jwt/v2 v2.3.0/go.mod h1:0tqz9Hlu6bCBFLWAASKhE5vUA4c24L9KPUUgvwumE/k=
-github.com/nats-io/nats-server/v2 v2.9.0 h1:DLWu+7/VgGOoChcDKytnUZPAmudpv7o/MhKmNrnH1RE=
-github.com/nats-io/nats-server/v2 v2.9.0/go.mod h1:BWKY6217RvhI+FDoOLZ2BH+hOC37xeKRBlQ1Lz7teKI=
-github.com/nats-io/nats.go v1.16.1-0.20220906180156-a1017eec10b0/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
+github.com/nats-io/nats-server/v2 v2.9.2 h1:XNDgJgOYYaYlquLdbSHI3xssLipfKUOq3EmYIMNCOsE=
+github.com/nats-io/nats-server/v2 v2.9.2/go.mod h1:4sq8wvrpbvSzL1n3ZfEYnH4qeUuIl5W990j3kw13rRk=
+github.com/nats-io/nats.go v1.17.0/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
 github.com/nats-io/nkeys v0.3.0 h1:cgM5tL53EvYRU+2YLXIK0G2mJtK12Ft9oeooSZMA2G8=
 github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV6A5y4=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
@@ -34,21 +34,21 @@ github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 go.uber.org/automaxprocs v1.5.1/go.mod h1:BF4eumQw0P9GtnuxxovUd06vwm1o18oMzFtK66vU6XU=
 golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
-golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90 h1:Y/gsMcFOcR+6S6f3YeMKl5g+dZMEWqcz5Czj/GWYbkM=
-golang.org/x/crypto v0.0.0-20220829220503-c86fa9a7ed90/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
+golang.org/x/crypto v0.0.0-20220926161630-eccd6366d1be h1:fmw3UbQh+nxngCAHrDCCztao/kbYFnWjoqop8dHx05A=
+golang.org/x/crypto v0.0.0-20220926161630-eccd6366d1be/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220906135438-9e1f76180b77 h1:C1tElbkWrsSkn3IRl1GCW/gETw1TywWIPgwZtXTZbYg=
-golang.org/x/sys v0.0.0-20220906135438-9e1f76180b77/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220928140112-f11e5e49a4ec h1:BkDtF2Ih9xZ7le9ndzTA7KJow28VbQW3odyk/8drmuI=
+golang.org/x/sys v0.0.0-20220928140112-f11e5e49a4ec/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/time v0.0.0-20220722155302-e5dcc9cfc0b9 h1:ftMN5LMiBFjbzleLqtoBZk7KdJwhuybIU+FckUHgoyQ=
-golang.org/x/time v0.0.0-20220722155302-e5dcc9cfc0b9/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/time v0.0.0-20220922220347-f3bd1da661af h1:Yx9k8YCG3dvF87UAn2tu2HQLf2dt/eR1bXxpLMWeH+Y=
+golang.org/x/time v0.0.0-20220922220347-f3bd1da661af/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
diff --git a/test/js_test.go b/test/js_test.go
index 8e4513e..ad1ac6a 100644
--- a/test/js_test.go
+++ b/test/js_test.go
@@ -8201,3 +8201,90 @@ func TestJetStreamOrderedConsumerRecreateAfterReconnect(t *testing.T) {
 		t.Fatalf("Invalid msg value; want: 'msg 2'; got: %q", string(msg.Data))
 	}
 }
+
+func TestJetStreamCreateStreamDiscardPolicy(t *testing.T) {
+	tests := []struct {
+		name                 string
+		discardPolicy        nats.DiscardPolicy
+		discardNewPerSubject bool
+		maxMsgsPerSubject    int64
+		withAPIError         bool
+	}{
+		{
+			name:                 "with discard policy 'new' and discard new per subject set",
+			discardPolicy:        nats.DiscardNew,
+			discardNewPerSubject: true,
+			maxMsgsPerSubject:    100,
+		},
+		{
+			name:                 "with discard policy 'new' and discard new per subject not set",
+			discardPolicy:        nats.DiscardNew,
+			discardNewPerSubject: false,
+			maxMsgsPerSubject:    100,
+		},
+		{
+			name:                 "with discard policy 'old' and discard new per subject set",
+			discardPolicy:        nats.DiscardOld,
+			discardNewPerSubject: true,
+			maxMsgsPerSubject:    100,
+			withAPIError:         true,
+		},
+		{
+			name:                 "with discard policy 'old' and discard new per subject not set",
+			discardPolicy:        nats.DiscardOld,
+			discardNewPerSubject: true,
+			maxMsgsPerSubject:    100,
+			withAPIError:         true,
+		},
+		{
+			name:                 "with discard policy 'new' and discard new per subject set and max msgs per subject not set",
+			discardPolicy:        nats.DiscardNew,
+			discardNewPerSubject: true,
+			withAPIError:         true,
+		},
+	}
+
+	s := RunBasicJetStreamServer()
+	defer shutdownJSServerAndRemoveStorage(t, s)
+
+	nc, js := jsClient(t, s)
+	defer nc.Close()
+
+	for i, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			streamName := fmt.Sprintf("FOO%d", i)
+
+			_, err := js.AddStream(&nats.StreamConfig{
+				Name:                 streamName,
+				Discard:              test.discardPolicy,
+				DiscardNewPerSubject: test.discardNewPerSubject,
+				MaxMsgsPerSubject:    test.maxMsgsPerSubject,
+			})
+
+			if test.withAPIError {
+				var apiErr *nats.APIError
+				if err == nil {
+					t.Fatalf("Expected error, got nil")
+				}
+				if ok := errors.As(err, &apiErr); !ok {
+					t.Fatalf("Expected nats.APIError, got %v", err)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+
+			info, err := js.StreamInfo(streamName)
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+			if info.Config.Discard != test.discardPolicy {
+				t.Fatalf("Invalid value of discard policy; want: %s; got: %s", test.discardPolicy.String(), info.Config.Discard.String())
+			}
+			if info.Config.DiscardNewPerSubject != test.discardNewPerSubject {
+				t.Fatalf("Invalid value of discard_new_per_subject; want: %t; got: %t", test.discardNewPerSubject, info.Config.DiscardNewPerSubject)
+			}
+		})
+	}
+}
diff --git a/test/kv_test.go b/test/kv_test.go
index 8531584..b7b7e9a 100644
--- a/test/kv_test.go
+++ b/test/kv_test.go
@@ -853,7 +853,7 @@ func TestListKeyValueStores(t *testing.T) {
 			if len(names) != test.bucketsNum {
 				t.Fatalf("Invalid number of stream names; want: %d; got: %d", test.bucketsNum, len(names))
 			}
-			infos := make([]nats.KeyValue, 0)
+			infos := make([]nats.KeyValueStatus, 0)
 			for info := range js.KeyValueStores() {
 				infos = append(infos, info)
 			}
diff --git a/test/object_test.go b/test/object_test.go
index 20b0f78..b9a1b30 100644
--- a/test/object_test.go
+++ b/test/object_test.go
@@ -893,7 +893,7 @@ func TestListObjectStores(t *testing.T) {
 			if len(names) != test.bucketsNum {
 				t.Fatalf("Invalid number of stream names; want: %d; got: %d", test.bucketsNum, len(names))
 			}
-			infos := make([]nats.ObjectStore, 0)
+			infos := make([]nats.ObjectStoreStatus, 0)
 			for info := range js.ObjectStores() {
 				infos = append(infos, info)
 			}