@@ -22,6 +22,7 @@ import (
2222 "path/filepath"
2323 "runtime"
2424 "sort"
25+ "strings"
2526 "sync"
2627 "time"
2728
@@ -33,7 +34,7 @@ import (
3334)
3435
3536type subCmd struct {
36- subject string
37+ subjects [] string
3738 queue string
3839 durable string
3940 raw bool
@@ -63,7 +64,7 @@ func configureSubCommand(app commandHost) {
6364 act := app .Command ("subscribe" , "Generic subscription client" ).Alias ("sub" ).Action (c .subscribe )
6465 addCheat ("sub" , act )
6566
66- act .Arg ("subject " , "Subject to subscribe to" ).StringVar (& c .subject )
67+ act .Arg ("subjects " , "Subjects to subscribe to" ).StringsVar (& c .subjects )
6768 act .Flag ("queue" , "Subscribe to a named queue group" ).StringVar (& c .queue )
6869 act .Flag ("durable" , "Use a durable consumer (requires JetStream)" ).StringVar (& c .durable )
6970 act .Flag ("raw" , "Show the raw data received" ).Short ('r' ).UnNegatableBoolVar (& c .raw )
@@ -159,15 +160,29 @@ func (c *subCmd) subscribe(p *fisk.ParseContext) error {
159160 }
160161 defer nc .Close ()
161162
162- if c .subject == "" && c .inbox {
163- c .subject = nc .NewRespInbox ()
164- } else if c .subject == "" && c .stream == "" {
163+ c .jetStream = c .sseq > 0 || len (c .durable ) > 0 || c .deliverAll || c .deliverNew || c .deliverLast || c .deliverSince != "" || c .deliverLastPerSubject || c .stream != ""
164+
165+ switch {
166+ case len (c .subjects ) == 0 && c .inbox :
167+ c .subjects = []string {nc .NewRespInbox ()}
168+ case len (c .subjects ) == 0 && c .stream == "" :
165169 return fmt .Errorf ("subject is required" )
170+ case len (c .subjects ) > 1 && c .jetStream :
171+ return fmt .Errorf ("streams subscribe support only 1 subject" )
166172 }
167173
174+ if c .inbox && c .jetStream {
175+ return fmt .Errorf ("generating inboxes is not compatible with JetStream subscriptions" )
176+ }
177+ if c .queue != "" && c .jetStream {
178+ return fmt .Errorf ("queue group subscriptions are not supported with JetStream" )
179+ }
168180 if c .dump == "-" && c .inbox {
169181 return fmt .Errorf ("generating inboxes is not compatible with dumping to stdout using null terminated strings" )
170182 }
183+ if c .reportSubjects && c .reportSubjectsCount == 0 {
184+ return fmt .Errorf ("subject count must be at least one" )
185+ }
171186
172187 if c .dump != "" && c .dump != "-" {
173188 err = os .MkdirAll (c .dump , 0700 )
@@ -176,21 +191,8 @@ func (c *subCmd) subscribe(p *fisk.ParseContext) error {
176191 }
177192 }
178193
179- c .jetStream = c .sseq > 0 || len (c .durable ) > 0 || c .deliverAll || c .deliverNew || c .deliverLast || c .deliverSince != "" || c .deliverLastPerSubject || c .stream != ""
180-
181- if c .inbox && c .jetStream {
182- return fmt .Errorf ("generating inboxes is not compatible with JetStream subscriptions" )
183- }
184- if c .queue != "" && c .jetStream {
185- return fmt .Errorf ("queue group subscriptions are not supported with JetStream" )
186- }
187-
188- if c .reportSubjects && c .reportSubjectsCount == 0 {
189- return fmt .Errorf ("subject count must be at least one" )
190- }
191-
192194 var (
193- sub * nats.Subscription
195+ subs [] * nats.Subscription
194196 mu = sync.Mutex {}
195197 subjMu = sync.Mutex {}
196198 dump = c .dump != ""
@@ -269,7 +271,10 @@ func (c *subCmd) subscribe(p *fisk.ParseContext) error {
269271 }
270272
271273 if ctr == c .limit {
272- sub .Unsubscribe ()
274+ for _ , sub := range subs {
275+ sub .Unsubscribe ()
276+ }
277+
273278 // if no reply matching, or if didn't yet get all replies
274279 if ! c .match || len (matchMap ) == 0 {
275280 cancel ()
@@ -335,15 +340,22 @@ func (c *subCmd) subscribe(p *fisk.ParseContext) error {
335340 case c .jetStream :
336341 // logs later depending on settings
337342 case c .jsAck :
338- log .Printf ("Subscribing on %s with acknowledgement of JetStream messages %s" , c .subject , ignoredSubjInfo )
343+ log .Printf ("Subscribing on %s with acknowledgement of JetStream messages %s" , c .firstSubject () , ignoredSubjInfo )
339344 default :
340- log .Printf ("Subscribing on %s %s" , c . subject , ignoredSubjInfo )
345+ log .Printf ("Subscribing on %s %s" , strings . Join ( c . subjects , ", " ) , ignoredSubjInfo )
341346 }
342347 }
343348
344349 switch {
345350 case c .reportSubjects :
346- sub , err = nc .Subscribe (c .subject , handler )
351+ for _ , subj := range c .subjects {
352+ sub , err := nc .Subscribe (subj , handler )
353+ if err != nil {
354+ return err
355+ }
356+ subs = append (subs , sub )
357+ }
358+
347359 startSubjectReporting (ctx , & subjMu , subjectReportMap , subjectBytesReportMap , c .reportSubjectsCount )
348360
349361 case c .jetStream :
@@ -378,9 +390,9 @@ func (c *subCmd) subscribe(p *fisk.ParseContext) error {
378390 }
379391 }
380392
381- subMsg := c .subject
393+ subMsg := c .firstSubject ()
382394 if c .stream != "" {
383- if c . subject == "" {
395+ if len ( c . subjects ) == 0 {
384396 str , err := js .StreamInfo (c .stream )
385397 if err != nil {
386398 return err
@@ -422,17 +434,36 @@ func (c *subCmd) subscribe(p *fisk.ParseContext) error {
422434 }
423435
424436 if bindDurable {
425- sub , err = js .Subscribe ("" , handler , nats .Bind (c .stream , c .durable ))
437+ sub , err := js .Subscribe ("" , handler , nats .Bind (c .stream , c .durable ))
438+ if err != nil {
439+ return err
440+ }
441+ subs = append (subs , sub )
426442 } else {
427443 c .jsAck = false
428- sub , err = js .Subscribe (c .subject , handler , opts ... )
444+ sub , err := js .Subscribe (c .firstSubject (), handler , opts ... )
445+ if err != nil {
446+ return err
447+ }
448+ subs = append (subs , sub )
429449 }
430450
431451 case c .queue != "" :
432- sub , err = nc .QueueSubscribe (c .subject , c .queue , handler )
452+ sub , err := nc .QueueSubscribe (c .firstSubject (), c .queue , handler )
453+ if err != nil {
454+ return err
455+ }
456+ subs = append (subs , sub )
433457
434458 default :
435- sub , err = nc .Subscribe (c .subject , handler )
459+ for _ , subj := range c .subjects {
460+ sub , err := nc .Subscribe (subj , handler )
461+ if err != nil {
462+ return err
463+ }
464+ subs = append (subs , sub )
465+ }
466+
436467 }
437468 if err != nil {
438469 return err
@@ -450,6 +481,14 @@ func (c *subCmd) subscribe(p *fisk.ParseContext) error {
450481 return nil
451482}
452483
484+ func (c * subCmd ) firstSubject () string {
485+ if len (c .subjects ) == 0 {
486+ return ""
487+ }
488+
489+ return c .subjects [0 ]
490+ }
491+
453492func (c * subCmd ) printMsg (msg * nats.Msg , reply * nats.Msg , ctr uint ) {
454493 var info * jsm.MsgInfo
455494 if msg .Reply != "" {
0 commit comments