/* I2PTunnel is GPL'ed (with the exception mentioned in I2PTunnel.java)
 * (c) 2003 - 2004 mihi
 */
package net.i2p.i2ptunnel;

import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InterruptedIOException;
import java.io.OutputStream;
import java.net.Socket;
import java.net.SocketException;
import java.util.HashMap;

import net.i2p.client.I2PSession;
import net.i2p.client.streaming.I2PSocket;
import net.i2p.util.Clock;
import net.i2p.util.I2PThread;
import net.i2p.util.Log;

public class I2PTunnelRunner extends I2PThread {
    private final static Log _log = new Log(I2PTunnelRunner.class);

    /** 
     * max bytes streamed in a packet - smaller ones might be filled
     * up to this size. Larger ones are not split (at least not on
     * Sun's impl of BufferedOutputStream), but that is the streaming
     * api's job...
     */
    static int MAX_PACKET_SIZE = 1024 * 32;

    static final int NETWORK_BUFFER_SIZE = MAX_PACKET_SIZE;

    private Socket s;
    private I2PSocket i2ps;
    Object slock, finishLock = new Object();
    boolean finished = false;
    HashMap ostreams, sockets;
    I2PSession session;
    byte[] initialData;
    /** when the last data was sent/received (or -1 if never) */
    private long lastActivityOn;
    /** when the runner started up */
    private long startedOn;

    public I2PTunnelRunner(Socket s, I2PSocket i2ps, Object slock, byte[] initialData) {
        this.s = s;
        this.i2ps = i2ps;
        this.slock = slock;
        this.initialData = initialData;
        lastActivityOn = -1;
        startedOn = -1;
        _log.info("I2PTunnelRunner started");
        setName("I2PTunnelRunner");
        start();
    }

    /** 
     * have we closed at least one (if not both) of the streams 
     * [aka we're done running the streams]? 
     *
     */
    public boolean isFinished() {
        return finished;
    }

    /** 
     * When was the last data for this runner sent or received?  
     *
     * @return date (ms since the epoch), or -1 if no data has been transferred yet
     *
     */
    public long getLastActivityOn() {
        return lastActivityOn;
    }

    private void updateActivity() {
        lastActivityOn = Clock.getInstance().now();
    }

    /**
     * When this runner started up transferring data
     *
     */
    public long getStartedOn() {
        return startedOn;
    }

    public void run() {
        startedOn = Clock.getInstance().now();
        try {
            InputStream in = s.getInputStream();
            OutputStream out = new BufferedOutputStream(s.getOutputStream(), NETWORK_BUFFER_SIZE);
            InputStream i2pin = i2ps.getInputStream();
            OutputStream i2pout = new BufferedOutputStream(i2ps.getOutputStream(), MAX_PACKET_SIZE);
            if (initialData != null) {
                synchronized (slock) {
                    i2pout.write(initialData);
                    i2pout.flush();
                }
            }
            Thread t1 = new StreamForwarder(in, i2pout);
            Thread t2 = new StreamForwarder(i2pin, out);
            synchronized (finishLock) {
                while (!finished) {
                    finishLock.wait();
                }
            }
            // now one connection is dead - kill the other as well.
            s.close();
            s = null;
            i2ps.close();
            i2ps = null;
            t1.join();
            t2.join();
        } catch (InterruptedException ex) {
            _log.error("Interrupted", ex);
        } catch (IOException ex) {
            ex.printStackTrace();
            _log.debug("Error forwarding", ex);
        } finally {
            try {
                if (s != null) s.close();
                if (i2ps != null) i2ps.close();
            } catch (IOException ex) {
                ex.printStackTrace();
                _log.error("Could not close socket", ex);
            }
        }
    }

    private class StreamForwarder extends I2PThread {

        InputStream in;
        OutputStream out;

        private StreamForwarder(InputStream in, OutputStream out) {
            this.in = in;
            this.out = out;
            setName("StreamForwarder");
            start();
        }

        public void run() {
            byte[] buffer = new byte[NETWORK_BUFFER_SIZE];
            try {
                int len;
                while ((len = in.read(buffer)) != -1) {
                    out.write(buffer, 0, len);

                    if (len > 0) updateActivity();

                    if (in.available() == 0) {
                        try {
                            Thread.sleep(I2PTunnel.PACKET_DELAY);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    if (in.available() == 0) {
                        out.flush(); // make sure the data get though
                    }
                }
            } catch (SocketException ex) {
                // this *will* occur when the other threads closes the socket
                synchronized (finishLock) {
                    if (!finished) {
                        _log.debug("Socket closed - error reading and writing",
                                   ex);
                    }
                }
            } catch (InterruptedIOException ex) {
                _log.warn("Closing connection due to timeout (error: \""
                          + ex.getMessage() + "\")");
            } catch (IOException ex) {
                if (!finished)
                    _log.error("Error forwarding", ex);
                else
                    _log.warn("You may ignore this", ex);
            } finally {
                try {
                    out.close();
                    in.close();
                } catch (IOException ex) {
                    _log.error("Error closing streams", ex);
                }
                synchronized (finishLock) {
                    finished = true;
                    finishLock.notifyAll();
                    // the main thread will close sockets etc. now
                }
            }
        }
    }
}

@@ -641,24 +641,97 @@
 er.class);%0A%0A
+    private static volatile long __runnerId;%0A    private long _runnerId;%0A
     /** %0A   
@@ -1734,24 +1734,58 @@
  started%22);%0A
+        _runnerId = ++__runnerId;%0A
         setN
@@ -1804,17 +1804,30 @@
 elRunner
-%22
+ %22 + _runnerId
 );%0A     
@@ -4083,32 +4083,82 @@
        %7D%0A    %7D%0A%0A
+    private volatile long __forwarderId = 0;%0A    %0A
     private clas
@@ -4405,17 +4405,56 @@
 orwarder
-%22
+ %22 + _runnerId + %22.%22 + (++__forwarderId)
 );%0A     

add a unique id to the thread names