Merge "Clean up & fix PipedInputStreamTest."

3 files changed, 46 insertions, 63 deletions

diff --git a/harmony-tests/src/test/java/org/apache/harmony/tests/java/io/PipedInputStreamTest.java b/harmony-tests/src/test/java/org/apache/harmony/tests/java/io/PipedInputStreamTest.java
index 9cf8054..6122dbb 100644
--- a/harmony-tests/src/test/java/org/apache/harmony/tests/java/io/PipedInputStreamTest.java
+++ b/harmony-tests/src/test/java/org/apache/harmony/tests/java/io/PipedInputStreamTest.java
@@ -19,6 +19,7 @@ package org.apache.harmony.tests.java.io;
 import java.io.IOException;
 import java.io.PipedInputStream;
 import java.io.PipedOutputStream;
+import java.util.concurrent.CountDownLatch;
 
 public class PipedInputStreamTest extends junit.framework.TestCase {
 
@@ -248,106 +249,83 @@ public class PipedInputStreamTest extends junit.framework.TestCase {
     /**
      * java.io.PipedInputStream#receive(int)
      */
-    public void test_receive() throws IOException {
+    public void test_write_failsAfterReaderDead() throws Exception {
         pis = new PipedInputStream();
         pos = new PipedOutputStream();
 
         // test if writer recognizes dead reader
         pis.connect(pos);
-        class WriteRunnable implements Runnable {
 
-            boolean pass = false;
+        class WriteRunnable implements Runnable {
 
-            volatile boolean readerAlive = true;
+            final CountDownLatch readerAlive = new CountDownLatch(1);
 
             public void run() {
                 try {
                     pos.write(1);
-                    while (readerAlive) {
-                        ;
+
+                    try {
+                        readerAlive.await();
+                    } catch (InterruptedException ie) {
+                        fail();
+                        return;
                     }
+
                     try {
                         // should throw exception since reader thread
                         // is now dead
                         pos.write(1);
-                    } catch (IOException e) {
-                        pass = true;
+                        fail();
+                    } catch (IOException expected) {
                     }
                 } catch (IOException e) {
                 }
             }
         }
-        WriteRunnable writeRunnable = new WriteRunnable();
-        Thread writeThread = new Thread(writeRunnable);
-        class ReadRunnable implements Runnable {
-
-            boolean pass;
 
+        class ReadRunnable implements Runnable {
             public void run() {
                 try {
                     pis.read();
-                    pass = true;
                 } catch (IOException e) {
+                    fail();
                 }
             }
         }
-        ;
+
+        WriteRunnable writeRunnable = new WriteRunnable();
+        Thread writeThread = new Thread(writeRunnable);
+
         ReadRunnable readRunnable = new ReadRunnable();
         Thread readThread = new Thread(readRunnable);
         writeThread.start();
         readThread.start();
-        while (readThread.isAlive()) {
-            ;
-        }
-        writeRunnable.readerAlive = false;
-        assertTrue("reader thread failed to read", readRunnable.pass);
-        while (writeThread.isAlive()) {
-            ;
-        }
-        assertTrue("writer thread failed to recognize dead reader",
-                writeRunnable.pass);
+        readThread.join();
 
-        // attempt to write to stream after writer closed
-        pis = new PipedInputStream();
-        pos = new PipedOutputStream();
+        writeRunnable.readerAlive.countDown();
+        writeThread.join();
+    }
 
-        pis.connect(pos);
-        class MyRunnable implements Runnable {
+    static final class PipedInputStreamWithPublicReceive extends PipedInputStream {
+        @Override
+        public void receive(int oneByte) throws IOException {
+            super.receive(oneByte);
+        }
+    }
 
-            boolean pass;
 
-            public void run() {
-                try {
-                    pos.write(1);
-                } catch (IOException e) {
-                    pass = true;
-                }
-            }
-        }
-        MyRunnable myRun = new MyRunnable();
-        synchronized (pis) {
-            t = new Thread(myRun);
-            // thread t will be blocked inside pos.write(1)
-            // when it tries to call the synchronized method pis.receive
-            // because we hold the monitor for object pis
-            t.start();
-            try {
-                // wait for thread t to get to the call to pis.receive
-                Thread.sleep(100);
-            } catch (InterruptedException e) {
-            }
-            // now we close
-            pos.close();
-        }
-        // we have exited the synchronized block, so now thread t will make
-        // a call to pis.receive AFTER the output stream was closed,
-        // in which case an IOException should be thrown
-        while (t.isAlive()) {
-            ;
+    public void test_receive_failsIfWriterClosed() throws Exception {
+        // attempt to write to stream after writer closed
+        PipedInputStreamWithPublicReceive pis = new PipedInputStreamWithPublicReceive();
+
+        pos = new PipedOutputStream();
+        pos.connect(pis);
+        pos.close();
+        try {
+            pis.receive(1);
+            fail();
+        } catch (IOException expected) {
         }
-        assertTrue(
-                "write failed to throw IOException on closed PipedOutputStream",
-                myRun.pass);
     }
 
     static class Worker extends Thread {
diff --git a/luni/src/main/java/java/io/PipedInputStream.java b/luni/src/main/java/java/io/PipedInputStream.java
index 2c27695..3e81cb8 100644
--- a/luni/src/main/java/java/io/PipedInputStream.java
+++ b/luni/src/main/java/java/io/PipedInputStream.java
@@ -390,6 +390,7 @@ public class PipedInputStream extends InputStream {
                 if (lastReader != null && !lastReader.isAlive()) {
                     throw new IOException("Pipe broken");
                 }
+
                 notifyAll();
                 wait(1000);
             }
@@ -402,6 +403,10 @@ public class PipedInputStream extends InputStream {
         if (in == -1) {
             in = 0;
         }
+        if (lastReader != null && !lastReader.isAlive()) {
+            throw new IOException("Pipe broken");
+        }
+
         buffer[in++] = (byte) oneByte;
         if (in == buffer.length) {
             in = 0;
diff --git a/luni/src/main/java/java/io/PipedOutputStream.java b/luni/src/main/java/java/io/PipedOutputStream.java
index 1b139e9..4c431c1 100644
--- a/luni/src/main/java/java/io/PipedOutputStream.java
+++ b/luni/src/main/java/java/io/PipedOutputStream.java
@@ -140,7 +140,7 @@ public class PipedOutputStream extends OutputStream {
      * @throws IOException
      *             if this stream is not connected, if the target stream is
      *             closed or if the thread reading from the target stream is no
-     *             longer alive. This case is currently not handled correctly.
+     *             longer alive.
      */
     @Override
     public void write(byte[] buffer, int offset, int count) throws IOException {