Repository: nats-io/natscli

Issue:
Support sending on newline for `nats pub`
### Proposed change

Provide a CLI flag for `nats pub <topic>` to tell it to send a message on every newline.

Example of sending on newline:

```sh
cat requests.jsonl | nats pub request --send-on-newline
```

Or if instead you wanted it more flexible for future use cases:

```sh
# What I would like it to do
cat requests.jsonl | nats pub request --send-on=newline
# The default and what it does today
cat requests.jsonl | nats pub request --send-on=eof
```

-----

Note: If this feature already exists or is possible somewhere, my apologies for the request. The idea is based off another's feedback + me skimming the docs and help options to see if something was missed. I didn't see an option there.

### Use case

I have a tool that can listen on stdin and publish on stdout in JSONL format.

It would be nice to be able to do:

```sh
nats sub request | my-command | nats pub response --send-on=newline
```

Or even for simple transforms like:

```sh
nats sub raw | jq -cr 'some-transform' | nats pub formatted --send-on=newline
```


### Contribution

Yep! If the proposal is accepted, I'd be up for me (or someone on my team) submitting a PR with the things needed to get the feature added ^_^

Hints from issue comments:
@ripienaar: I think it's a good idea, would be happy to review.  Would need to be careful and support it for both JetStream and non JetStream publishing.@bplaxco: >  Would need to be careful and support it for both JetStream and non JetStream publishing.

Okay, thanks for that note. ^_^@alexbozhenko: Does it have to be part of the natscli?

I feel like one can achieve that with [xargs](https://www.man7.org/linux/man-pages/man1/xargs.1.html)?
```
cat /tmp/jsonl.txt | xargs -I % nats pub subject "%"
```

It has this flag:
```
       --delimiter=delim, -d delim
              Input items are terminated by the specified character. ...
```
That would allow split input on any char.
@thewizzy: > Does it have to be part of the natscli?
> 
> I feel like one can achieve that with [xargs](https://www.man7.org/linux/man-pages/man1/xargs.1.html)?
> 
> ```
> cat /tmp/jsonl.txt | xargs -I % nats pub subject "%"
> ```
> 
> It has this flag:
> 
> ```
>        --delimiter=delim, -d delim
>               Input items are terminated by the specified character. ...
> ```
> 
> That would allow split input on any char.

It is correct, this could be achieved with xargs, though it could catch delimiters (like newlines \n) in the body. 

More significantly is that every line of the file executes`nats`, connects to the server and then cleans up the process. 

I did a simple test to see how significant the different was, the server was located on the same local network but a different machine.
**xargs - using release 0.2.2**
`time cat nats/tests/consumer_command_test.go | xargs -I % -d \n nats pub testing -s 192.168.1.2 "%"`
```
real    0m30.511s
user    0m5.517s
sys     0m4.088s
```
Note: to be fair the delimiter here does mean that the xargs method published more lines.

**send-on newline**
`time cat nats/tests/consumer_command_test.go |  nats pub testing -s 192.168.1.2 --send-on newline`
```
real    0m0.814s
user    0m0.032s
sys     0m0.114s
```@ripienaar: Done in https://github.com/nats-io/natscli/pull/1356

Solution patch:
diff --git a/cli/cheats/pub.md b/cli/cheats/pub.md
index 65b1038..1d8f4a8 100644
--- a/cli/cheats/pub.md
+++ b/cli/cheats/pub.md
@@ -9,3 +9,6 @@ echo "hello world" | nats pub --force-stdin destination.subject
 
 # To request a response from a server and show just the raw result
 nats request destination.subject "hello world" -H "Content-type:text/plain" --raw
+
+# To listen on STDIN and publish one message per newline
+nats pub destination.subject --send-on=newline
\ No newline at end of file
diff --git a/cli/pub_command.go b/cli/pub_command.go
index d688fec..db82afc 100644
--- a/cli/pub_command.go
+++ b/cli/pub_command.go
@@ -14,10 +14,14 @@
 package cli
 
 import (
+	"bufio"
+	"bytes"
 	"fmt"
 	"io"
 	"math"
 	"os"
+	"os/signal"
+	"syscall"
 	"time"
 
 	iu "github.com/nats-io/natscli/internal/util"
@@ -29,6 +33,11 @@ import (
 	terminal "golang.org/x/term"
 )
 
+const (
+	sendOnEOF     = "eof"
+	sendOnNewline = "newline"
+)
+
 type pubCmd struct {
 	subject      string
 	body         string
@@ -44,6 +53,8 @@ type pubCmd struct {
 	forceStdin   bool
 	translate    string
 	jetstream    bool
+	sendOn       string
+	quiet        bool
 }
 
 func configurePubCommand(app commandHost) {
@@ -80,6 +91,9 @@ Available template functions are:
 	pub.Flag("sleep", "When publishing multiple messages, sleep between publishes").DurationVar(&c.sleep)
 	pub.Flag("force-stdin", "Force reading from stdin").UnNegatableBoolVar(&c.forceStdin)
 	pub.Flag("jetstream", "Publish messages to jetstream").Short('J').UnNegatableBoolVar(&c.jetstream)
+	pub.Flag("send-on", fmt.Sprintf("When to send data from stdin: '%s' (default) or '%s'", sendOnEOF, sendOnNewline)).
+		Default("eof").EnumVar(&c.sendOn, sendOnNewline, sendOnEOF)
+	pub.Flag("quiet", "Show just the output received").Short('q').UnNegatableBoolVar(&c.quiet)
 
 	requestHelp := `Body and Header values of the messages may use Go templates to 
 create unique messages.
@@ -204,7 +218,7 @@ func (c *pubCmd) doReq(nc *nats.Conn, progress *progress.Tracker) error {
 							log.Printf("%s: %s", h, val)
 						}
 					}
-					fmt.Println()
+					log.Println()
 				}
 
 				outPutMSGBody(m.Data, c.translate, m.Subject, "")
@@ -259,9 +273,9 @@ func (c *pubCmd) doJetstream(nc *nats.Conn, progress *progress.Tracker) error {
 		}
 
 		msg.Subject = string(subj)
-
-		log.Printf("Published %d bytes to %q\n", len(body), c.subject)
-
+		if !c.quiet {
+			log.Printf("Published %d bytes to %q\n", len(body), c.subject)
+		}
 		resp, err := nc.RequestMsg(msg, opts().Timeout)
 		if err != nil {
 			return err
@@ -278,7 +292,7 @@ func (c *pubCmd) doJetstream(nc *nats.Conn, progress *progress.Tracker) error {
 
 		if progress != nil {
 			progress.Increment(1)
-		} else {
+		} else if !c.quiet {
 			msg := fmt.Sprintf("Stored in Stream: %s Sequence: %s", ack.Stream, f(ack.Sequence))
 			if ack.Domain != "" {
 				msg += fmt.Sprintf(" Domain: %q", ack.Domain)
@@ -301,88 +315,180 @@ func (c *pubCmd) doJetstream(nc *nats.Conn, progress *progress.Tracker) error {
 	return nil
 }
 
+// readLine reads a full line from a bufio.Reader regardless of buffer size,
+// this is important for reading lines longer than the bufio.Reader's default
+// buffer size and nicer than wholesale enlarging the buffer.
+func readLine(reader *bufio.Reader) (string, error) {
+	var buf bytes.Buffer
+
+	for {
+		line, isPrefix, err := reader.ReadLine()
+		if err != nil {
+			return buf.String(), err
+		}
+
+		buf.Write(line)
+		if !isPrefix {
+			return buf.String(), err
+		}
+	}
+}
+
 func (c *pubCmd) publish(_ *fisk.ParseContext) error {
+	ctx, cancel := signal.NotifyContext(ctx, syscall.SIGTERM, syscall.SIGINT, syscall.SIGQUIT)
+	defer cancel()
+
 	nc, err := newNatsConn("", natsOpts()...)
 	if err != nil {
 		return err
 	}
 	defer nc.Close()
 
-	if c.cnt < 1 {
-		c.cnt = math.MaxInt16
-	}
+	// Create a pipe so that we can interrupt the input at any time.
+	readPipe, writePipe := io.Pipe()
+	go func() {
+		_, errPipe := io.Copy(writePipe, os.Stdin)
+		if errPipe != nil {
+			writePipe.CloseWithError(errPipe)
+		} else {
+			// io.Copy does not return io.EOF, so on success we need to trigger EOF that by closing the pipe
+			_ = writePipe.Close()
+		}
+	}()
+	reader := bufio.NewReader(readPipe)
+	complete := make(chan struct{})
 
-	if !c.bodyIsSet && (terminal.IsTerminal(int(os.Stdout.Fd())) || c.forceStdin) {
+	// If a body is set, treat it as EOF, as no more input
+	eof := c.bodyIsSet
+	if !c.bodyIsSet && !c.quiet && (terminal.IsTerminal(int(os.Stdout.Fd())) || c.forceStdin) {
 		log.Println("Reading payload from STDIN")
-		body, err := io.ReadAll(os.Stdin)
-		if err != nil {
-			return err
-		}
-		c.body = string(body)
 	}
 
-	var tracker *progress.Tracker
-	var progbar progress.Writer
+	errCh := make(chan error, 1)
+	go func() {
+		defer close(complete)
+		for {
+			if c.cnt < 1 {
+				c.cnt = math.MaxInt16
+			}
 
-	if c.cnt > 20 && !c.raw {
-		progbar, tracker, err = iu.NewProgress(opts(), &progress.Tracker{
-			Total: int64(c.cnt),
-		})
-		if err != nil {
-			return err
-		}
+			if !c.bodyIsSet && (terminal.IsTerminal(int(os.Stdout.Fd())) || c.forceStdin) {
+				if c.sendOn == sendOnEOF {
+					body, err := io.ReadAll(readPipe)
+					if err != nil {
+						errCh <- err
+						return
+					}
+					c.body = string(body)
+				} else if c.sendOn == sendOnNewline {
+					body, err := readLine(reader)
+					if err != nil && err != io.EOF {
+						errCh <- err
+						return
+					} else if err == io.EOF {
+						eof = true
+					}
+					if body == "" && eof {
+						errCh <- nil
+						return
+					}
+					c.body = body
+				}
+			}
 
-		defer func() {
-			progbar.Stop()
-			time.Sleep(300 * time.Millisecond)
-		}()
-	}
+			var tracker *progress.Tracker
+			var progbar progress.Writer
 
-	if c.jetstream {
-		return c.doJetstream(nc, tracker)
-	}
+			if c.cnt > 20 && !c.raw {
+				progbar, tracker, err = iu.NewProgress(opts(), &progress.Tracker{
+					Total: int64(c.cnt),
+				})
+				if err != nil {
+					errCh <- err
+					return
+				}
 
-	if c.req || c.replyCount >= 1 {
-		return c.doReq(nc, tracker)
-	}
+				defer func() {
+					progbar.Stop()
+					time.Sleep(300 * time.Millisecond)
+				}()
+			}
 
-	for i := 1; i <= c.cnt; i++ {
-		body, err := iu.PubReplyBodyTemplate(c.body, "", i)
-		if err != nil {
-			log.Printf("Could not parse body template: %s", err)
-		}
+			if c.jetstream {
+				err = c.doJetstream(nc, tracker)
+				if c.sendOn == sendOnEOF {
+					errCh <- err
+					return
+				} else if c.sendOn == sendOnNewline {
+					if err != nil {
+						log.Printf("Could not publish message: %s", err)
+					}
+					continue
+				}
+			}
 
-		subj, err := iu.PubReplyBodyTemplate(c.subject, "", i)
-		if err != nil {
-			log.Printf("Could not parse subject template: %s", err)
-		}
+			if c.req || c.replyCount >= 1 {
+				errCh <- c.doReq(nc, tracker)
+				return
+			}
 
-		msg, err := c.prepareMsg(string(subj), body, i)
-		if err != nil {
-			return err
-		}
+			for i := 1; i <= c.cnt; i++ {
+				body, err := iu.PubReplyBodyTemplate(c.body, "", i)
+				if err != nil {
+					log.Printf("Could not parse body template: %s", err)
+				}
 
-		err = nc.PublishMsg(msg)
-		if err != nil {
-			return err
-		}
-		nc.Flush()
+				subj, err := iu.PubReplyBodyTemplate(c.subject, "", i)
+				if err != nil {
+					log.Printf("Could not parse subject template: %s", err)
+				}
 
-		err = nc.LastError()
-		if err != nil {
-			return err
-		}
+				msg, err := c.prepareMsg(string(subj), body, i)
+				if err != nil {
+					errCh <- err
+					return
+				}
 
-		if c.cnt > 1 && c.sleep > 0 {
-			time.Sleep(c.sleep)
-		}
+				err = nc.PublishMsg(msg)
+				if err != nil {
+					errCh <- err
+					return
+				}
+				nc.Flush()
 
-		if progbar == nil {
-			log.Printf("Published %d bytes to %q\n", len(body), c.subject)
-		} else {
-			tracker.Increment(1)
+				err = nc.LastError()
+				if err != nil {
+					errCh <- err
+					return
+				}
+
+				if c.cnt > 1 && c.sleep > 0 {
+					time.Sleep(c.sleep)
+				}
+
+				if progbar == nil {
+					if !c.quiet {
+						log.Printf("Published %d bytes to %q\n", len(body), c.subject)
+					}
+				} else {
+					tracker.Increment(1)
+				}
+			}
+
+			if c.sendOn == sendOnEOF || eof {
+				errCh <- nil
+				return
+			}
 		}
+	}()
+
+	// Wait until the core go routine is complete or context is canceled.
+	// Closes the remote connection(due to defer), the core loop go routine is just dropped.
+	select {
+	case <-ctx.Done():
+		readPipe.Close()
+		return fmt.Errorf("interrupted")
+	case <-complete:
+		return <-errCh
 	}
-
-	return nil
 }



Test patch:
diff --git a/nats/tests/pub_command_test.go b/nats/tests/pub_command_test.go
new file mode 100644
index 0000000..7d17762
--- /dev/null
+++ b/nats/tests/pub_command_test.go
@@ -0,0 +1,97 @@
+package main
+
+import (
+	"fmt"
+	"github.com/nats-io/nats.go"
+	"strings"
+	"testing"
+)
+
+func TestCLIPubSendOnNewline(t *testing.T) {
+	srv, _, _ := setupJStreamTest(t)
+	defer srv.Shutdown()
+	nc, _, _ := prepareHelper(srv.ClientURL())
+
+	t.Run("Publish with body argument", func(t *testing.T) {
+		subject := "test-body"
+		var messages []string
+		expected := "Test Message"
+		sub, _ := nc.Subscribe(subject, func(m *nats.Msg) {
+			messages = append(messages, string(m.Data))
+		})
+
+		runNatsCli(t, fmt.Sprintf("--server='%s' pub --send-on newline %s '%s'", srv.ClientURL(), subject, expected))
+		_ = sub.Unsubscribe()
+
+		if len(messages) != 1 {
+			t.Errorf("expected 1 message and received %d", len(messages))
+		}
+		if len(messages) > 0 && messages[0] != expected {
+			t.Errorf("expected message %q got %q", expected, messages[0])
+		}
+	})
+
+	t.Run("Publish with body argument and --quiet", func(t *testing.T) {
+		subject := "test-quiet"
+		var messages []string
+		expected := "Test Message"
+		sub, _ := nc.Subscribe(subject, func(m *nats.Msg) {
+			messages = append(messages, string(m.Data))
+		})
+
+		out := runNatsCli(t, fmt.Sprintf("--server='%s' pub --send-on newline -q %s '%s'", srv.ClientURL(), subject, expected))
+		_ = sub.Unsubscribe()
+
+		if len(out) != 0 {
+			t.Errorf("expected cli to output to be 0 but got %d\n %s", len(out), out)
+		}
+		if len(messages) != 1 {
+			t.Errorf("expected 1 message and received %d", len(messages))
+		}
+		if len(messages) > 0 && messages[0] != expected {
+			t.Errorf("expected message %q got %q", expected, messages[0])
+		}
+	})
+
+	t.Run("Publish --send-on newline from stdin", func(t *testing.T) {
+		subject := "test-sendon"
+		var messages []string
+		expected := []string{"test", "pub", "input"}
+		sub, _ := nc.Subscribe(subject, func(m *nats.Msg) {
+			messages = append(messages, string(m.Data))
+		})
+
+		// --force-stdin required for testing as a terminal is not present
+		runNatsCliWithInput(t, strings.Join(expected, "\n"), fmt.Sprintf("--server='%s' pub --send-on newline --force-stdin %s", srv.ClientURL(), subject))
+		_ = sub.Unsubscribe()
+
+		if len(messages) != len(expected) {
+			t.Errorf("expected %d messages and received %d", len(expected), len(messages))
+		}
+		for i, msg := range messages {
+			if messages[i] != expected[i] {
+				t.Errorf("expected message(%d) %q got %q", i, expected[i], msg)
+			}
+		}
+	})
+
+	t.Run("Publish --send-on eof from stdin", func(t *testing.T) {
+		subject := "test-eof"
+		var messages []string
+		expected := "test\npub\ninput"
+		sub, _ := nc.Subscribe(subject, func(m *nats.Msg) {
+			messages = append(messages, string(m.Data))
+		})
+
+		// --force-stdin required for testing as a terminal is not present
+		runNatsCliWithInput(t, expected, fmt.Sprintf("--server='%s' pub --force-stdin %s", srv.ClientURL(), subject))
+		_ = sub.Unsubscribe()
+
+		if len(messages) != 1 {
+			t.Errorf("expected 1 message and received %d", len(messages))
+		}
+		if len(messages) > 0 && messages[0] != expected {
+			t.Errorf("expected message %q got %q", expected, messages[0])
+		}
+	})
+}