From 9bc5022c9c0916b2c7f6be8dfcaafcfb30f2122f Mon Sep 17 00:00:00 2001
From: Jonathan Turner <jonathandturner@users.noreply.github.com>
Date: Fri, 14 Feb 2020 18:15:32 -0800
Subject: [PATCH] Force a \n at the end of a stdout stream (#1391)

* Force a \n at the end of a stdout stream

* clippy
---
 src/cli.rs                          |  4 ++-
 src/commands/classified/external.rs | 41 ++++++++++++++++++++++++++++-
 2 files changed, 43 insertions(+), 2 deletions(-)

diff --git a/src/cli.rs b/src/cli.rs
index 2ffe1f4e57..341d634727 100644
--- a/src/cli.rs
+++ b/src/cli.rs
@@ -635,7 +635,9 @@ async fn process_line(
             }
 
             let input_stream = if redirect_stdin {
-                let file = futures::io::AllowStdIo::new(std::io::stdin());
+                let file = futures::io::AllowStdIo::new(
+                    crate::commands::classified::external::StdoutWithNewline::new(std::io::stdin()),
+                );
                 let stream = FramedRead::new(file, LinesCodec).map(|line| {
                     if let Ok(line) = line {
                         Ok(Value {
diff --git a/src/commands/classified/external.rs b/src/commands/classified/external.rs
index 33c275d7f4..a798117e9e 100644
--- a/src/commands/classified/external.rs
+++ b/src/commands/classified/external.rs
@@ -379,6 +379,45 @@ async fn run_with_stdin(
     spawn(&command, &path, &process_args[..], input, is_last)
 }
 
+/// This is a wrapper for stdout-like readers that ensure a carriage return ends the stream
+pub struct StdoutWithNewline<T: std::io::Read> {
+    stdout: T,
+    ended_in_newline: bool,
+}
+
+impl<T: std::io::Read> StdoutWithNewline<T> {
+    pub fn new(stdout: T) -> StdoutWithNewline<T> {
+        StdoutWithNewline {
+            stdout,
+            ended_in_newline: false,
+        }
+    }
+}
+impl<T: std::io::Read> std::io::Read for StdoutWithNewline<T> {
+    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+        match self.stdout.read(buf) {
+            Err(e) => Err(e),
+            Ok(0) => {
+                if !self.ended_in_newline && !buf.is_empty() {
+                    self.ended_in_newline = true;
+                    buf[0] = b'\n';
+                    Ok(1)
+                } else {
+                    Ok(0)
+                }
+            }
+            Ok(len) => {
+                if buf[len - 1] == b'\n' {
+                    self.ended_in_newline = true;
+                } else {
+                    self.ended_in_newline = false;
+                }
+                Ok(len)
+            }
+        }
+    }
+}
+
 fn spawn(
     command: &ExternalCommand,
     path: &str,
@@ -477,7 +516,7 @@ fn spawn(
                     return;
                 };
 
-                let file = futures::io::AllowStdIo::new(stdout);
+                let file = futures::io::AllowStdIo::new(StdoutWithNewline::new(stdout));
                 let mut stream = FramedRead::new(file, LinesCodec);
 
                 while let Some(line) = stream.next().await {