Repository: nats-io/nats.go
Pull request #1090 (2022-09-20 07:58:58)

[CHANGED] Remove restriction on AckNone policy for pull consumers
Addresses https://github.com/nats-io/nats-architecture-and-design/issues/156

coveralls: 
[![Coverage Status](https://coveralls.io/builds/52604285/badge)](https://coveralls.io/builds/52604285)

Coverage increased (+0.006%) to 85.775% when pulling **917b744944d2b622653be9997e140734f51b21cb on ack-none-pull-consumer** into **715a5917c806d2858f6fc80663fbf0910ad58ee3 on main**.


Patch:
diff --git a/js.go b/js.go
index 22ad5f3..4cbc282 100644
--- a/js.go
+++ b/js.go
@@ -1420,10 +1420,6 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,
 
 	// Some checks for pull subscribers
 	if isPullMode {
-		// Check for bad ack policy
-		if o.cfg.AckPolicy == AckNonePolicy {
-			return nil, fmt.Errorf("nats: invalid ack mode for pull consumers: %s", o.cfg.AckPolicy)
-		}
 		// No deliver subject should be provided
 		if o.cfg.DeliverSubject != _EMPTY_ {
 			return nil, ErrPullSubscribeToPushConsumer
diff --git a/kv.go b/kv.go
index 2c74848..a25485c 100644
--- a/kv.go
+++ b/kv.go
@@ -35,6 +35,10 @@ type KeyValueManager interface {
 	CreateKeyValue(cfg *KeyValueConfig) (KeyValue, error)
 	// DeleteKeyValue will delete this KeyValue store (JetStream stream).
 	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
 }
 
 // Notice: Experimental Preview
@@ -325,16 +329,7 @@ func (js *js) KeyValue(bucket string) (KeyValue, error) {
 		return nil, ErrBadBucket
 	}
 
-	kv := &kvs{
-		name:   bucket,
-		stream: stream,
-		pre:    fmt.Sprintf(kvSubjectsPreTmpl, bucket),
-		js:     js,
-		// Determine if we need to use the JS prefix in front of Put and Delete operations
-		useJSPfx:  js.opts.pre != defaultAPIPrefix,
-		useDirect: si.Config.AllowDirect,
-	}
-	return kv, nil
+	return mapStreamToKVS(js, si), nil
 }
 
 // CreateKeyValue will create a KeyValue store with the following configuration.
@@ -431,17 +426,7 @@ func (js *js) CreateKeyValue(cfg *KeyValueConfig) (KeyValue, error) {
 			return nil, err
 		}
 	}
-
-	kv := &kvs{
-		name:   cfg.Bucket,
-		stream: scfg.Name,
-		pre:    fmt.Sprintf(kvSubjectsPreTmpl, cfg.Bucket),
-		js:     js,
-		// Determine if we need to use the JS prefix in front of Put and Delete operations
-		useJSPfx:  js.opts.pre != defaultAPIPrefix,
-		useDirect: si.Config.AllowDirect,
-	}
-	return kv, nil
+	return mapStreamToKVS(js, si), nil
 }
 
 // DeleteKeyValue will delete this KeyValue store (JetStream stream).
@@ -980,3 +965,55 @@ func (kv *kvs) Status() (KeyValueStatus, error) {
 
 	return &KeyValueBucketStatus{nfo: nfo, bucket: kv.name}, nil
 }
+
+// KeyValueStoreNames is used to retrieve a list of key value store names
+func (js *js) KeyValueStoreNames() <-chan string {
+	ch := make(chan string)
+	l := &streamLister{js: js}
+	l.js.opts.streamListSubject = fmt.Sprintf(kvSubjectsTmpl, "*")
+	go func() {
+		defer close(ch)
+		for l.Next() {
+			for _, info := range l.Page() {
+				if !strings.HasPrefix(info.Config.Name, "KV_") {
+					continue
+				}
+				ch <- info.Config.Name
+			}
+		}
+	}()
+
+	return ch
+}
+
+// KeyValueStores is used to retrieve a list of key value stores
+func (js *js) KeyValueStores() <-chan KeyValue {
+	ch := make(chan KeyValue)
+	l := &streamLister{js: js}
+	l.js.opts.streamListSubject = fmt.Sprintf(kvSubjectsTmpl, "*")
+	go func() {
+		defer close(ch)
+		for l.Next() {
+			for _, info := range l.Page() {
+				if !strings.HasPrefix(info.Config.Name, "KV_") {
+					continue
+				}
+				ch <- mapStreamToKVS(js, info)
+			}
+		}
+	}()
+	return ch
+}
+
+func mapStreamToKVS(js *js, info *StreamInfo) *kvs {
+	bucket := strings.TrimPrefix(info.Config.Name, "KV_")
+	return &kvs{
+		name:   bucket,
+		stream: info.Config.Name,
+		pre:    fmt.Sprintf(kvSubjectsPreTmpl, bucket),
+		js:     js,
+		// Determine if we need to use the JS prefix in front of Put and Delete operations
+		useJSPfx:  js.opts.pre != defaultAPIPrefix,
+		useDirect: info.Config.AllowDirect,
+	}
+}


Test patch:
diff --git a/test/js_test.go b/test/js_test.go
index e2a448e..e5ad87d 100644
--- a/test/js_test.go
+++ b/test/js_test.go
@@ -686,6 +686,14 @@ func TestJetStreamSubscribe(t *testing.T) {
 		t.Fatalf("Error on subscribe: %v", err)
 	}
 	sub.Unsubscribe()
+
+	// Pull consumer with AckNone policy
+	sub, err = js.PullSubscribe("bar", "", nats.AckNone())
+	if err != nil {
+		t.Fatalf("Error on subscribe: %v", err)
+	}
+	sub.Unsubscribe()
+
 	// Can't specify DeliverSubject for pull subscribers
 	_, err = js.PullSubscribe("bar", "foo", nats.DeliverSubject("baz"))
 	if err != nats.ErrPullSubscribeToPushConsumer {
diff --git a/test/kv_test.go b/test/kv_test.go
index 4ece985..8531584 100644
--- a/test/kv_test.go
+++ b/test/kv_test.go
@@ -813,3 +813,53 @@ func expectErr(t *testing.T, err error, expected ...error) {
 	}
 	t.Fatalf("Expected one of %+v, got '%v'", expected, err)
 }
+
+func TestListKeyValueStores(t *testing.T) {
+	tests := []struct {
+		name       string
+		bucketsNum int
+	}{
+		{
+			name:       "single page",
+			bucketsNum: 5,
+		},
+		{
+			name:       "multi page",
+			bucketsNum: 1025,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			s := RunBasicJetStreamServer()
+			defer shutdownJSServerAndRemoveStorage(t, s)
+
+			nc, js := jsClient(t, s)
+			defer nc.Close()
+			// create stream without the chunk subject, but with KV_ prefix
+			_, err := js.AddStream(&nats.StreamConfig{Name: "KV_FOO", Subjects: []string{"FOO.*"}})
+			expectOk(t, err)
+			// create stream with chunk subject, but without "KV_" prefix
+			_, err = js.AddStream(&nats.StreamConfig{Name: "FOO", Subjects: []string{"$KV.ABC.>"}})
+			expectOk(t, err)
+			for i := 0; i < test.bucketsNum; i++ {
+				_, err = js.CreateKeyValue(&nats.KeyValueConfig{Bucket: fmt.Sprintf("KVS_%d", i), MaxBytes: 1024})
+				expectOk(t, err)
+			}
+			names := make([]string, 0)
+			for name := range js.KeyValueStoreNames() {
+				names = append(names, name)
+			}
+			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)
+			for info := range js.KeyValueStores() {
+				infos = append(infos, info)
+			}
+			if len(infos) != test.bucketsNum {
+				t.Fatalf("Invalid number of streams; want: %d; got: %d", test.bucketsNum, len(infos))
+			}
+		})
+	}
+}
diff --git a/test/object_test.go b/test/object_test.go
index bbede20..f1447b0 100644
--- a/test/object_test.go
+++ b/test/object_test.go
@@ -791,7 +791,7 @@ func TestObjectMaxBytes(t *testing.T) {
 	}
 }
 
-func TestBucketNames(t *testing.T) {
+func TestListObjectStores(t *testing.T) {
 	tests := []struct {
 		name       string
 		bucketsNum int