Merge aa63249 into a3519e1

js.go

@@ -168,10 +168,27 @@ func (opt jsOptFn) configureJSContext(opts *js) error {
 
 func APIPrefix(pre string) JSOpt {
 	return jsOptFn(func(js *js) error {
-		js.pre = pre
-		if !strings.HasSuffix(js.pre, ".") {
-			js.pre = js.pre + "."
+		pre = strings.TrimSuffix(pre, ".")
+
+		if pre == ">" {
+			return ErrJetStreamBadPre
 		}
+
+		toks := strings.Split(pre, ".")
+		for i, tok := range toks {
+			if tok == "" || tok == "*" {
+				return ErrJetStreamBadPre
+			}
+
+			if tok == ">" && i < len(toks)-1 {
+				return ErrBadSubject
+			} else if tok == ">" && i == len(toks)-1 {
+				js.pre = pre
+				return nil
+			}
+		}
+
+		js.pre = fmt.Sprintf("%s.", pre)
 		return nil
 	})
 }
@@ -617,7 +634,6 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, opts []
 
 func (js *js) lookupStreamBySubject(subj string) (string, error) {
 	var slr JSApiStreamNamesResponse
-	// FIXME(dlc) - prefix
 	req := &streamRequest{subj}
 	j, err := json.Marshal(req)
 	if err != nil {
@@ -734,7 +750,6 @@ func (sub *Subscription) Poll() error {
 }
 
 func (js *js) getConsumerInfo(stream, consumer string) (*ConsumerInfo, error) {
-	// FIXME(dlc) - prefix
 	ccInfoSubj := fmt.Sprintf(JSApiConsumerInfoT, stream, consumer)
 	resp, err := js.nc.Request(js.apiSubj(ccInfoSubj), nil, js.wait)
 	if err != nil {

nats_test.go

@@ -2599,3 +2599,61 @@ func TestMsg_RespondMsg(t *testing.T) {
 		t.Fatalf("did not get correct response: %q", resp.Data)
 	}
 }
+
+func TestJetStreamAPIPrefix(t *testing.T) {
+	cases := []struct {
+		prefix     string
+		wantPrefix string
+		wantErr    bool
+	}{
+		{
+			prefix:     "foo",
+			wantPrefix: "foo.",
+		},
+		{
+			prefix:     "foo.",
+			wantPrefix: "foo.",
+		},
+		{
+			prefix:     "foo.>",
+			wantPrefix: "foo.>",
+		},
+		{
+			prefix:     "foo.b*r.baz",
+			wantPrefix: "foo.b*r.baz.",
+		},
+		{
+			prefix:  "foo.*",
+			wantErr: true,
+		},
+		{
+			prefix:  "foo.*.bar",
+			wantErr: true,
+		},
+		{
+			prefix:  "foo.>.bar",
+			wantErr: true,
+		},
+		{
+			prefix:  ">",
+			wantErr: true,
+		},
+	}
+	for _, c := range cases {
+		t.Run(c.prefix, func(t *testing.T) {
+			jsOpt := APIPrefix(c.prefix)
+
+			js := new(js)
+			if err := jsOpt.configureJSContext(js); err != nil && !c.wantErr {
+				t.Fatal(err)
+			} else if err == nil && c.wantErr {
+				t.Fatal("unexpected success")
+			}
+
+			if js.pre != c.wantPrefix {
+				t.Error("unexpected api prefix")
+				t.Fatalf("got=%s; want=%s", js.pre, c.wantPrefix)
+			}
+		})
+	}
+}