package com.hypirion.io;

import java.io.InputStream;
import java.io.OutputStream;

public class Pipe {

    public static final int DEFAULT_BUFFER_SIZE = 1024;

    private final Thread threadPumper;
    private final PipeThread pt;
    private final InputStream in;
    private final OutputStream out;
    private final int bufsize;
    private final Object dataLock;

    public Pipe(InputStream in, OutputStream out) {
        this(in, out, DEFAULT_BUFFER_SIZE);
    }

    public Pipe(InputStream in, OutputStream out, int bufsize) {
        this.in = in;
        this.out = out;
        this.bufsize = bufsize;
        pt = new PipeThread();
        threadPumper = new Thread(pt);
        threadPumper.setName(String.format("PipeThread %s %s", in.hashCode(),
                                           out.hashCode()));
        threadPumper.setDaemon(true);
        dataLock = new Object();
    }

    public void join() throws InterruptedException {
        threadPumper.join();
    }

    public void start() {

    }

    public void stop() {
        stop(true);
    }

    public void stop(boolean block) {
        
    }

    private class PipeThread implements Runnable {
        private final byte[] data;

        public PipeThread() {
            data = new byte[bufsize];
        }

        @Override
        public void run() {
            while (true) {
                synchronized (dataLock) {
                    int count = in.read(data);
                    if (count < 0) {
                        out.close();
                        break;
                    }
                    out.write(data, 0, count);
                }
            }
        }
    }
}

@@ -75,16 +75,60 @@
 tStream;
+%0Aimport java.util.concurrent.CountDownLatch;
 %0A%0Apublic
@@ -397,24 +397,67 @@
  Object 
-dataLock
+lock;%0A    private volatile boolean currentlyRunning
 ;%0A%0A    p
@@ -956,31 +956,91 @@
     
-dataLock = new Object()
+threadPumper.start();%0A        lock = new Object();%0A        currentlyRunning = false
 ;%0A  
@@ -1140,24 +1140,37 @@
 %0A    public 
+synchronized 
 void start()
@@ -1172,16 +1172,158 @@
 art() %7B%0A
+        synchronized (lock) %7B%0A            currentlyRunning = true;%0A            lock.notify(); // Wake up the pumper if it's waiting.%0A        %7D
 %0A    %7D%0A%0A
@@ -1701,27 +1701,141 @@
 ed (
-dataLock) %7B%0A       
+lock) %7B%0A                    while (!currentlyRunning) %7B%0A                        lock.wait();%0A                    %7D%0A                %7D%0A
     
@@ -1843,16 +1843,19 @@
         
+   
 int coun
@@ -1889,20 +1889,16 @@
         
-    
 if (coun
@@ -1902,28 +1902,24 @@
 ount %3C 0) %7B%0A
-    
             
@@ -1959,20 +1959,16 @@
         
-    
 break;%0A 
@@ -1978,32 +1978,26 @@
             
-    %7D%0A  
+%7D%0A
             
@@ -1996,26 +1996,24 @@
             
-  
 out.write(da
@@ -2027,34 +2027,16 @@
 count);%0A
-                %7D%0A
         

Implement thread-safe start of Pipe.