Add publish request wait
@@ -36,6 +36,7 @@ type pubCmd struct {
hdrs []string
cnt int
sleep time.Duration
+ waitDur time.Duration
forceStdin bool
}
@@ -67,6 +68,7 @@ Available template functions are:
pub.Arg("subject", "Subject to subscribe to").Required().StringVar(&c.subject)
pub.Arg("body", "Message body").Default("!nil!").StringVar(&c.body)
pub.Flag("wait", "Wait for a reply from a service").Short('w').BoolVar(&c.req)
+ pub.Flag("wait-duration", "Wait for multiple replies from a service").Short('d').Default("2s").DurationVar(&c.waitDur)
pub.Flag("reply", "Sets a custom reply to subject").StringVar(&c.replyTo)
pub.Flag("header", "Adds headers to the message").Short('H').StringsVar(&c.hdrs)
pub.Flag("count", "Publish multiple messages").Default("1").IntVar(&c.cnt)
@@ -129,6 +131,73 @@ func (c *pubCmd) prepareMsg(body []byte, seq int) (*nats.Msg, error) {
return msg, parseStringsToMsgHeader(c.hdrs, seq, msg)
}
+func (c *pubCmd) doReqWait(nc *nats.Conn, progress *uiprogress.Bar) error {
+ for i := 1; i <= c.cnt; i++ {
+ start := time.Now()
+
+ if !c.raw && progress == nil {
+ log.Printf("Sending request on %q\n", c.subject)
+ }
+
+ body, err := pubReplyBodyTemplate(c.body, 0)
+ if err != nil {
+ log.Printf("Could not parse body template: %s", err)
+ }
+
+ msg, err := c.prepareMsg(body, 0)
+ if err != nil {
+ return err
+ }
+ msg.Reply = nc.NewRespInbox()
+
+ s, err := nc.Subscribe(msg.Reply, func(m *nats.Msg) {
+ log.Printf("Received on %q rtt %v", m.Subject, time.Since(start))
+
+ if len(m.Header) > 0 {
+ for h, vals := range m.Header {
+ for _, val := range vals {
+ log.Printf("%s: %s", h, val)
+ }
+ }
+
+ fmt.Println()
+ }
+
+ fmt.Println(string(m.Data))
+ if !strings.HasSuffix(string(m.Data), "\n") {
+ fmt.Println()
+ }
+ })
+ if err != nil {
+ return err
+ }
+ defer s.Unsubscribe()
+
+ err = nc.PublishMsg(msg)
+ if err != nil {
+ return err
+ }
+
+ // cls - TODO maybe add response count and select with
+ // channel as well to short circuit if response count
+ // is known apriori.
+ time.Sleep(c.waitDur)
+
+ // we don't want any responses after we wait.
+ s.Unsubscribe()
+
+ // include a publish sleep in the wait duration
+ if c.cnt > 1 && c.sleep > 0 {
+ st := c.sleep - c.waitDur
+ if st > 0 {
+ time.Sleep(st)
+ }
+ }
+ }
+
+ return nil
+}
+
func (c *pubCmd) doReq(nc *nats.Conn, progress *uiprogress.Bar) error {
for i := 1; i <= c.cnt; i++ {
start := time.Now()
@@ -214,6 +283,10 @@ func (c *pubCmd) publish(_ *kingpin.ParseContext) error {
defer func() { uiprogress.Stop(); fmt.Println() }()
}
+ if c.waitDur != 0 {
+ return c.doReqWait(nc, progress)
+ }
+
if c.req {
return c.doReq(nc, progress)
}
