Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Make lookupStreamBySubject public #1114

Merged
merged 8 commits into from Oct 31, 2022
Merged

Make lookupStreamBySubject public #1114

Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
e4df87d
Make lookupStreamBySubject public
Jarema Oct 28, 2022
e34d8e8
Add StreamBySubject to the jsm interface
Jarema Oct 28, 2022
96f3833
Add test prefix
Jarema Oct 28, 2022
de487d5
Move StreanNameBySubject to jsm.go
Jarema Oct 28, 2022
ac3ef92
Refactor test into test table
Jarema Oct 28, 2022
06dad43
Get rid of newline
Jarema Oct 28, 2022
bff7b16
Revert go.mod and go.sum
Jarema Oct 28, 2022
a1ff370
Add JSOpt to StreamNameBySubject
Jarema Oct 31, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
26 changes: 1 addition & 25 deletions js.go
View file
Open in desktop
Expand Up @@ -1529,7 +1529,7 @@ func (js *js) subscribe(subj, queue string, cb MsgHandler, ch chan *Msg, isSync,

// Find the stream mapped to the subject if not bound to a stream already.
if o.stream == _EMPTY_ {
stream, err = js.lookupStreamBySubject(subj)
stream, err = js.StreamNameBySubject(subj)
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -2144,30 +2144,6 @@ type streamNamesResponse struct {
Streams []string `json:"streams"`
}

func (js *js) lookupStreamBySubject(subj string) (string, error) {
var slr streamNamesResponse
req := &streamRequest{subj}
j, err := json.Marshal(req)
if err != nil {
return _EMPTY_, err
}
resp, err := js.nc.Request(js.apiSubj(apiStreams), j, js.opts.wait)
if err != nil {
if err == ErrNoResponders {
err = ErrJetStreamNotEnabled
}
return _EMPTY_, err
}
if err := json.Unmarshal(resp.Data, &slr); err != nil {
return _EMPTY_, err
}

if slr.Error != nil || len(slr.Streams) != 1 {
return _EMPTY_, ErrNoMatchingStream
}
return slr.Streams[0], nil
}

type subOpts struct {
// For attaching.
stream, consumer string
Expand Down
39 changes: 39 additions & 0 deletions js_test.go
View file
Open in desktop
Expand Up @@ -1230,3 +1230,42 @@ func TestJetStreamStreamInfoWithSubjectDetails(t *testing.T) {
t.Fatalf("expected 0 subjects details from StreamInfo, but got %d instead", len(result.State.Subjects))
}
}

func TestStreamNameBySubject(t *testing.T) {
s := RunBasicJetStreamServer()
defer shutdownJSServerAndRemoveStorage(t, s)

nc, js := jsClient(t, s)
defer nc.Close()

var err error

_, err = js.AddStream(&StreamConfig{
Name: "TEST",
Subjects: []string{"test.*"},
})
if err != nil {
t.Fatalf("Unexpected error: %v", err)
}

for _, test := range []struct {
name string
streamName string
err error
}{

{name: "valid wildcard lookup", streamName: "test.*", err: nil},
{name: "valid explicit lookup", streamName: "test.a", err: nil},
{name: "lookup on not existing stream", streamName: "not.existing", err: ErrNoMatchingStream},
} {

stream, err := js.StreamNameBySubject(test.streamName)
if err != test.err {
t.Fatalf("expected %v, got %v", test.err, err)
}

if stream != "TEST" && err == nil {
t.Fatalf("returned stream name should be 'TEST'")
}
}
}
28 changes: 28 additions & 0 deletions jsm.go
View file
Open in desktop
Expand Up @@ -93,6 +93,9 @@ type JetStreamManager interface {

// AccountInfo retrieves info about the JetStream usage from an account.
AccountInfo(opts ...JSOpt) (*AccountInfo, error)

// StreamNameBySubjec rteturns a stream matching given subject.
StreamNameBySubject(string) (string, error)
}

// StreamConfig will determine the properties for a stream.
Expand Down Expand Up @@ -1515,6 +1518,31 @@ func (jsc *js) StreamNames(opts ...JSOpt) <-chan string {
return ch
}

// StreamNameBySubject returns a stream name that matches the subject.
func (jsc *js) StreamNameBySubject(subj string) (string, error) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am not sure this is right. Since this becomes a context level API, should you pass the js options as a vararg? Then inside you would call getJSContextOpts(), etc... User should be able to override the wait and apiSubj() still need fixing to make sure it uses any user domain provided override (granted not part of this PR).

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

While I'm not a huge fan of the concept of having 1 common option type for all interface methods, in this case I agree. It makes sense to allow user to at least pass context/wait and be consistent with the rest of the interface at the same time.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

agree. Updated the signature, and used jsc.apiRequestWithContext to at least use the context that can be passed, but still, it's misleading to give the user option to pass Domain and ApiPrefix and not use it.
That's consistent and I agree to have it, but it only adds up to what #1104 refers to.

var slr streamNamesResponse
req := &streamRequest{subj}
j, err := json.Marshal(req)
if err != nil {
return _EMPTY_, err
}
resp, err := jsc.nc.Request(jsc.apiSubj(apiStreams), j, jsc.opts.wait)
if err != nil {
if err == ErrNoResponders {
err = ErrJetStreamNotEnabled
}
return _EMPTY_, err
}
if err := json.Unmarshal(resp.Data, &slr); err != nil {
return _EMPTY_, err
}

if slr.Error != nil || len(slr.Streams) != 1 {
return _EMPTY_, ErrNoMatchingStream
}
return slr.Streams[0], nil
}

func getJSContextOpts(defs *jsOpts, opts ...JSOpt) (*jsOpts, context.CancelFunc, error) {
var o jsOpts
for _, opt := range opts {
Expand Down