support jetstream in basic sub command
@@ -17,7 +17,9 @@ import (
"context"
"fmt"
"log"
+ "strings"
+ "github.com/nats-io/jsm.go"
"github.com/nats-io/nats.go"
"gopkg.in/alecthomas/kingpin.v2"
)
@@ -26,6 +28,7 @@ type subCmd struct {
subject string
queue string
raw bool
+ jsAck bool
}
func configureSubCommand(app *kingpin.Application) {
@@ -34,6 +37,7 @@ func configureSubCommand(app *kingpin.Application) {
act.Arg("subject", "Subject to subscribe to").Required().StringVar(&c.subject)
act.Flag("queue", "Subscribe to a named queue group").StringVar(&c.queue)
act.Flag("raw", "Show the raw data received").Short('r').BoolVar(&c.raw)
+ act.Flag("ack", "Acknowledge JetStream message that have the correct metadata").BoolVar(&c.jsAck)
}
func (c *subCmd) subscribe(_ *kingpin.ParseContext) error {
@@ -52,6 +56,28 @@ func (c *subCmd) subscribe(_ *kingpin.ParseContext) error {
return
}
+ var info *jsm.MsgInfo
+ if m.Reply != "" {
+ info, _ = jsm.ParseJSMsgMetadata(m)
+ }
+
+ if info != nil {
+ log.Printf("[#%d] Received JetStream message: consumer: %s > %s / subject: %s / delivered: %d / consumer seq: %d / stream seq: %d / ack: %v\n\n", i, info.Stream(), info.Consumer(), m.Subject, info.Delivered(), info.ConsumerSequence(), info.StreamSequence(), c.jsAck)
+ fmt.Println(string(m.Data))
+ if !strings.HasSuffix(string(m.Data), "\n") {
+ fmt.Println()
+ }
+
+ if c.jsAck {
+ err = m.Respond(nil)
+ if err != nil {
+ fmt.Printf("Acknowledging message via subject %s failed: %s\n", m.Reply, err)
+ }
+ }
+
+ return
+ }
+
log.Printf("[#%d] Received on [%s]: '%s'", i, m.Subject, string(m.Data))
}
@@ -68,7 +94,12 @@ func (c *subCmd) subscribe(_ *kingpin.ParseContext) error {
}
if !c.raw {
- log.Printf("Listening on [%s]", c.subject)
+ if c.jsAck {
+ log.Printf("Subscribing on %s with acknowledgement of JetStream messages\n", c.subject)
+ } else {
+ log.Printf("Subscribing on %s\n", c.subject)
+ }
+
}
<-context.Background().Done()
