simplify BlockingProcessStreamReader; fix tests

google-cloud-core/src/main/java/com/google/cloud/testing/BlockingProcessStreamReader.java

@@ -35,15 +35,13 @@
  */
 class BlockingProcessStreamReader extends Thread {
 
-  private static final int STREAM_READER_SLEEP_INTERVAL_IN_MS = 200;
   private static final int LOG_LENGTH_LIMIT = 50000;
 
   private final BufferedReader errorReader;
   private final Logger logger;
   private StringBuilder currentLog;
   private Level currentLogLevel;
   private boolean collectionMode;
-  private volatile boolean terminated;
   private final String emulatorTag;
   private final Pattern logLinePattern;
 
@@ -64,34 +62,25 @@ private BlockingProcessStreamReader(String emulator, InputStream stream, String
   }
 
   void terminate() throws IOException {
-    terminated = true;
-    errorReader.close();
     interrupt();
   }
 
   @Override
   public void run() {
     String previousLine = "";
     String nextLine = "";
-    while (!terminated) {
-      try {
-        if (errorReader.ready()) {
-          previousLine = nextLine;
-          nextLine = errorReader.readLine();
-          if (nextLine == null) {
-            terminated = true;
-          } else {
-            processLogLine(previousLine, nextLine);
-          }
-        } else {
-          sleep(STREAM_READER_SLEEP_INTERVAL_IN_MS);
+    try {
+      for (;;) {
+        previousLine = nextLine;
+        nextLine = errorReader.readLine();
+        if (nextLine == null) {
+          break;
         }
-      } catch (IOException e) {
+        processLogLine(previousLine, nextLine);
+      }
+    } catch (IOException e) {
+      if (!isInterrupted()) {
         e.printStackTrace(System.err);
-      } catch (InterruptedException e) {
-        previousLine = nextLine;
-        nextLine = null;
-        break;
       }
     }
     processLogLine(previousLine, firstNonNull(nextLine, ""));

google-cloud-core/src/test/java/com/google/cloud/testing/BlockingProcessStreamReaderTest.java

@@ -87,17 +87,9 @@ public void testBlockUntil() throws IOException {
   public void testForwardLogEntry() throws IOException, InterruptedException {
     TestLogger logger = new TestLogger();
     InputStream stream = new ByteArrayInputStream(OUTPUT_WITH_LOGS.getBytes(Charsets.UTF_8));
-    BlockingProcessStreamReader thread =
-        BlockingProcessStreamReader.start("emulator", stream, BLOCK_UNTIL, logger);
-    while (logger.getLogs().get(Level.INFO).isEmpty()) {
-      Thread.sleep(200);
-    }
+    BlockingProcessStreamReader.start("emulator", stream, BLOCK_UNTIL, logger).join();
     assertEquals("[emulator] log line 1" + System.lineSeparator() + "[emulator] log line 2",
         logger.getLogs().get(Level.INFO).iterator().next());
-    thread.terminate();
-    while (logger.getLogs().get(Level.FINE).isEmpty()) {
-      Thread.sleep(200);
-    }
     assertEquals("[emulator] log line 3", logger.getLogs().get(Level.FINE).iterator().next());
     stream.close();
   }
@@ -106,17 +98,9 @@ public void testForwardLogEntry() throws IOException, InterruptedException {
   public void testForwardAlreadyTaggedLogs() throws IOException, InterruptedException {
     TestLogger logger = new TestLogger();
     InputStream stream = new ByteArrayInputStream(TAGGED_OUTPUT_WITH_LOGS.getBytes(Charsets.UTF_8));
-    BlockingProcessStreamReader thread =
-        BlockingProcessStreamReader.start("emulator", stream, BLOCK_UNTIL, logger);
-    while (logger.getLogs().get(Level.INFO).isEmpty()) {
-      Thread.sleep(200);
-    }
+    BlockingProcessStreamReader.start("emulator", stream, BLOCK_UNTIL, logger).join();
     assertEquals("[emulator] log line 1" + System.lineSeparator() + "[emulator] log line 2",
         logger.getLogs().get(Level.INFO).iterator().next());
-    thread.terminate();
-    while (logger.getLogs().get(Level.FINE).isEmpty()) {
-      Thread.sleep(200);
-    }
     assertEquals("[emulator] log line 3", logger.getLogs().get(Level.FINE).iterator().next());
     stream.close();
   }