Move streaming demo out of tests

apps/ministreaming/java/demo/net/i2p/client/streaming/StreamSinkClient.java

@@ -0,0 +1,187 @@
+package net.i2p.client.streaming;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.io.OutputStream;
+import java.net.ConnectException;
+import java.net.NoRouteToHostException;
+import java.util.Properties;
+import java.util.Random;
+
+import net.i2p.I2PAppContext;
+import net.i2p.I2PException;
+import net.i2p.data.DataFormatException;
+import net.i2p.data.Destination;
+import net.i2p.util.I2PThread;
+import net.i2p.util.Log;
+
+/**
+ * Simple streaming lib test app that connects to a given destination and sends 
+ * it a particular amount of random data, then disconnects.
+ * @see #main(java.lang.String[])
+ */
+public class StreamSinkClient {
+    private Log _log;
+    private int _sendSize;
+    private int _writeDelay;
+    private String _peerDestFile;
+    private String _i2cpHost;
+    private int _i2cpPort;
+
+    
+    /**
+     * Build the client but don't fire it up.
+     * @param sendSize how many KB to send
+     * @param writeDelayMs how long to wait between each .write (0 for no delay)
+     * @param serverDestFile file containing the StreamSinkServer's binary Destination
+     */
+    public StreamSinkClient(int sendSize, int writeDelayMs, String serverDestFile) {
+        this(null, -1, sendSize, writeDelayMs, serverDestFile);
+    }
+    public StreamSinkClient(String i2cpHost, int i2cpPort, int sendSize, int writeDelayMs, String serverDestFile) {
+        _i2cpHost = i2cpHost;
+        _i2cpPort = i2cpPort;
+        _sendSize = sendSize;
+        _writeDelay = writeDelayMs;
+        _peerDestFile = serverDestFile;
+        _log = I2PAppContext.getGlobalContext().logManager().getLog(StreamSinkClient.class);
+    }
+    
+    /**
+     * Actually connect and run the client - this call blocks until completion.
+     *
+     */
+    public void runClient() {
+        I2PSocketManager mgr = null;
+        if (_i2cpHost != null)
+            mgr = I2PSocketManagerFactory.createManager(_i2cpHost, _i2cpPort, new Properties());
+        else
+            mgr = I2PSocketManagerFactory.createManager();
+        Destination peer = null;
+        FileInputStream fis = null;
+        try { 
+            fis = new FileInputStream(_peerDestFile);
+            peer = new Destination();
+            peer.readBytes(fis);
+        } catch (IOException ioe) {
+            _log.error("Error finding the peer destination to contact in " + _peerDestFile, ioe);
+            return;
+        } catch (DataFormatException dfe) {
+            _log.error("Peer destination is not valid in " + _peerDestFile, dfe);
+            return;
+        } finally {
+            if (fis != null) try { fis.close(); } catch (IOException ioe) {}
+        }
+
+        if (_log.shouldLog(Log.DEBUG))
+            _log.debug("Send " + _sendSize + "KB to " + peer.calculateHash().toBase64());
+
+        while (true) {
+            try {
+                I2PSocket sock = mgr.connect(peer);
+                byte buf[] = new byte[Math.min(32*1024, _sendSize*1024)];
+                Random rand = new Random();
+                OutputStream out = sock.getOutputStream();
+                long beforeSending = System.currentTimeMillis();
+                for (int i = 0; (_sendSize < 0) || (i < _sendSize); i+= buf.length/1024) {
+                    rand.nextBytes(buf);
+                    out.write(buf);
+                    if (_log.shouldLog(Log.DEBUG))
+                        _log.debug("Wrote " + ((1+i*buf.length)/1024) + "/" + _sendSize + "KB");
+                    if (_writeDelay > 0) {
+                        try { Thread.sleep(_writeDelay); } catch (InterruptedException ie) {}
+                    }   
+                }
+                sock.close();
+                long afterSending = System.currentTimeMillis();
+                if (_log.shouldLog(Log.DEBUG))
+                    _log.debug("Sent " + _sendSize + "KB in " + (afterSending-beforeSending) + "ms");
+            } catch (InterruptedIOException iie) {
+                _log.error("Timeout connecting to the peer", iie);
+                //return;
+            } catch (NoRouteToHostException nrthe) {
+                _log.error("Unable to connect to the peer", nrthe);
+                //return;
+            } catch (ConnectException ce) {
+                _log.error("Connection already dropped", ce);
+                //return;
+            } catch (I2PException ie) {
+                _log.error("Error connecting to the peer", ie);
+                return;
+            } catch (IOException ioe) {
+                _log.error("IO error sending", ioe);
+                return;
+            }   
+        }
+    }
+
+    /**
+     * Fire up the client.  <code>Usage: StreamSinkClient [i2cpHost i2cpPort] sendSizeKB writeDelayMs serverDestFile [concurrentSends]</code> <br>
+     * <ul>
+     *  <li><b>sendSizeKB</b>: how many KB to send, or -1 for unlimited</li>
+     *  <li><b>writeDelayMs</b>: how long to wait between each .write (0 for no delay)</li>
+     *  <li><b>serverDestFile</b>: file containing the StreamSinkServer's binary Destination</li>
+     *  <li><b>concurrentSends</b>: how many concurrent threads should send to the server at once</li>
+     * </ul>
+     * @param args [i2cpHost i2cpPort] sendSizeKB writeDelayMs serverDestFile [concurrentSends]
+     */
+    public static void main(String args[]) {
+        StreamSinkClient client = null;
+        int sendSizeKB = -1;
+        int writeDelayMs = -1;
+        int concurrent = 1;
+        
+        switch (args.length) {
+            case 3: // fall through
+            case 4:
+                try {
+                    sendSizeKB = Integer.parseInt(args[0]);
+                } catch (NumberFormatException nfe) {
+                    System.err.println("Send size invalid [" + args[0] + "]");
+                    return;
+                }
+                try {
+                    writeDelayMs = Integer.parseInt(args[1]);
+                } catch (NumberFormatException nfe) {
+                    System.err.println("Write delay ms invalid [" + args[1] + "]");
+                    return;
+                }
+                if (args.length == 4) {
+                    try { concurrent = Integer.parseInt(args[3]); } catch (NumberFormatException nfe) {}
+                }
+                client = new StreamSinkClient(sendSizeKB, writeDelayMs, args[2]);
+                break;
+            case 5: // fall through
+            case 6:
+                try { 
+                    int port = Integer.parseInt(args[1]);
+                    sendSizeKB = Integer.parseInt(args[2]);
+                    writeDelayMs = Integer.parseInt(args[3]);
+                    client = new StreamSinkClient(args[0], port, sendSizeKB, writeDelayMs, args[4]);
+                } catch (NumberFormatException nfe) {
+                    System.err.println("arg error");
+                }
+                if (args.length == 6) {
+                    try { concurrent = Integer.parseInt(args[5]); } catch (NumberFormatException nfe) {}
+                }
+                break;
+            default: 
+                System.out.println("Usage: StreamSinkClient [i2cpHost i2cpPort] sendSizeKB writeDelayMs serverDestFile [concurrentSends]");
+        }
+        if (client != null) {
+            for (int i = 0; i < concurrent; i++)
+                new I2PThread(new Runner(client), "Client " + i).start();
+        }   
+    }
+    
+    private static class Runner implements Runnable {
+        private StreamSinkClient _client;
+        public Runner(StreamSinkClient client) {
+            _client = client;
+        }
+        public void run() {
+            _client.runClient();
+        }
+    }
+}

apps/ministreaming/java/demo/net/i2p/client/streaming/StreamSinkSend.java

@@ -0,0 +1,130 @@
+package net.i2p.client.streaming;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.io.OutputStream;
+import java.net.ConnectException;
+import java.net.NoRouteToHostException;
+
+import net.i2p.I2PAppContext;
+import net.i2p.I2PException;
+import net.i2p.data.DataFormatException;
+import net.i2p.data.Destination;
+import net.i2p.util.Log;
+
+/**
+ * Simple streaming lib test app that connects to a given destination and sends 
+ * the contents of a file, then disconnects.  See the {@link #main}
+ *
+ */
+public class StreamSinkSend {
+    private Log _log;
+    private String _sendFile;
+    private int _writeDelay;
+    private String _peerDestFile;
+    
+    /**
+     * Build the client but don't fire it up.
+     * @param filename file to send
+     * @param writeDelayMs how long to wait between each .write (0 for no delay)
+     * @param serverDestFile file containing the StreamSinkServer's binary Destination
+     */
+    public StreamSinkSend(String filename, int writeDelayMs, String serverDestFile) {
+        _sendFile = filename;
+        _writeDelay = writeDelayMs;
+        _peerDestFile = serverDestFile;
+        _log = I2PAppContext.getGlobalContext().logManager().getLog(StreamSinkClient.class);
+    }
+    
+    /**
+     * Actually connect and run the client - this call blocks until completion.
+     *
+     */
+    public void runClient() {
+        I2PSocketManager mgr = I2PSocketManagerFactory.createManager();
+        Destination peer = null;
+        FileInputStream fis = null;
+        try { 
+            fis = new FileInputStream(_peerDestFile);
+            peer = new Destination();
+            peer.readBytes(fis);
+        } catch (IOException ioe) {
+            _log.error("Error finding the peer destination to contact in " + _peerDestFile, ioe);
+            return;
+        } catch (DataFormatException dfe) {
+            _log.error("Peer destination is not valid in " + _peerDestFile, dfe);
+            return;
+        } finally {
+            if (fis != null) try { fis.close(); } catch (IOException ioe) {}
+        }
+        
+        
+        System.out.println("Send " + _sendFile + " to " + peer.calculateHash().toBase64());
+        
+        try {
+            I2PSocket sock = mgr.connect(peer);
+            byte buf[] = new byte[32*1024];
+            OutputStream out = sock.getOutputStream();
+            long beforeSending = System.currentTimeMillis();
+            fis = new FileInputStream(_sendFile);
+            long size = 0;
+            while (true) {
+                int read = fis.read(buf);
+                if (read < 0)
+                    break;
+                out.write(buf, 0, read);
+                size += read;
+                if (_log.shouldLog(Log.DEBUG))
+                    _log.debug("Wrote " + read);
+                if (_writeDelay > 0) {
+                    try { Thread.sleep(_writeDelay); } catch (InterruptedException ie) {}
+                }   
+            }
+            fis.close();
+            sock.close();
+            long afterSending = System.currentTimeMillis();
+            System.out.println("Sent " + (size / 1024) + "KB in " + (afterSending-beforeSending) + "ms");
+        } catch (InterruptedIOException iie) {
+            _log.error("Timeout connecting to the peer", iie);
+            return;
+        } catch (NoRouteToHostException nrthe) {
+            _log.error("Unable to connect to the peer", nrthe);
+            return;
+        } catch (ConnectException ce) {
+            _log.error("Connection already dropped", ce);
+            return;
+        } catch (I2PException ie) {
+            _log.error("Error connecting to the peer", ie);
+            return;
+        } catch (IOException ioe) {
+            _log.error("IO error sending", ioe);
+            return;
+        }
+    }
+
+    /**
+     * Fire up the client.  <code>Usage: StreamSinkClient sendFile writeDelayMs serverDestFile</code> <br>
+     * <ul>
+     *  <li><b>sendFile</b>: filename to send</li>
+     *  <li><b>writeDelayMs</b>: how long to wait between each .write (0 for no delay)</li>
+     *  <li><b>serverDestFile</b>: file containing the StreamSinkServer's binary Destination</li>
+     * </ul>
+     * @param args sendFile writeDelayMs serverDestFile
+     */
+    public static void main(String args[]) {
+        if (args.length != 3) {
+            System.out.println("Usage: StreamSinkClient sendFile writeDelayMs serverDestFile");
+        } else {
+            int writeDelayMs = -1;
+            try {
+                writeDelayMs = Integer.parseInt(args[1]);
+            } catch (NumberFormatException nfe) {
+                System.err.println("Write delay ms invalid [" + args[1] + "]");
+                return;
+            }
+            StreamSinkSend client = new StreamSinkSend(args[0], writeDelayMs, args[2]);
+            client.runClient();
+        }
+    }
+}

apps/ministreaming/java/demo/net/i2p/client/streaming/StreamSinkServer.java

@@ -0,0 +1,196 @@
+package net.i2p.client.streaming;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.ConnectException;
+import java.net.SocketTimeoutException;
+import java.util.Properties;
+
+import net.i2p.I2PAppContext;
+import net.i2p.I2PException;
+import net.i2p.data.DataFormatException;
+import net.i2p.data.Destination;
+import net.i2p.util.I2PThread;
+import net.i2p.util.Log;
+
+/**
+ * Listen to a destination, receiving any sockets and writing anything they 
+ * send to a new file. See the {@link #main}
+ *
+ */
+public class StreamSinkServer {
+    private Log _log;
+    private String _sinkDir;
+    private String _destFile;
+    private String _i2cpHost;
+    private int _i2cpPort;
+    private int _handlers;
+    
+    /**
+     * Create but do not start the streaming server.  
+     *
+     * @param sinkDir Directory to store received files in
+     * @param ourDestFile filename to write our binary destination to
+     */
+    public StreamSinkServer(String sinkDir, String ourDestFile) {
+        this(sinkDir, ourDestFile, null, -1, 3);
+    }
+    public StreamSinkServer(String sinkDir, String ourDestFile, String i2cpHost, int i2cpPort, int handlers) {
+        _sinkDir = sinkDir;
+        _destFile = ourDestFile;
+        _i2cpHost = i2cpHost;
+        _i2cpPort = i2cpPort;
+        _handlers = handlers;
+        _log = I2PAppContext.getGlobalContext().logManager().getLog(StreamSinkServer.class);
+    }
+    
+    /**
+     * Actually fire up the server - this call blocks forever (or until the server 
+     * socket closes)
+     * 
+     */
+    public void runServer() {
+        I2PSocketManager mgr = null;
+        if (_i2cpHost != null)
+            mgr = I2PSocketManagerFactory.createManager(_i2cpHost, _i2cpPort, new Properties());
+        else 
+            mgr = I2PSocketManagerFactory.createManager();
+        Destination dest = mgr.getSession().getMyDestination();
+        if (_log.shouldLog(Log.INFO))
+            _log.info("Listening for connections on: " + dest.calculateHash().toBase64());
+        FileOutputStream fos = null;
+        try {
+            fos = new FileOutputStream(_destFile);
+            dest.writeBytes(fos);
+        } catch (IOException ioe) {
+            _log.error("Error writing out our destination to " + _destFile, ioe);
+            return;
+        } catch (DataFormatException dfe) {
+            _log.error("Error formatting the destination", dfe);
+            return;
+        } finally {
+            if (fos != null) try { fos.close(); } catch (IOException ioe) {}
+        }
+        
+        I2PServerSocket sock = mgr.getServerSocket();
+        startup(sock);
+    }
+    
+    public void startup(I2PServerSocket sock) {
+        for (int i = 0; i < _handlers; i++) {
+            I2PThread t = new I2PThread(new ClientRunner(sock));
+            t.setName("Handler " + i);
+            t.setDaemon(false);
+            t.start();
+        }
+    }
+    
+    /**
+     * Actually deal with a client - pull anything they send us and write it to a file.
+     *
+     */
+    private class ClientRunner implements Runnable {
+        private I2PServerSocket _socket;
+        public ClientRunner(I2PServerSocket socket) {
+            _socket = socket;
+        }
+        public void run() {
+            while (true) {
+                try {
+                    I2PSocket socket = _socket.accept();
+                    if (socket != null)
+                        handle(socket);
+                } catch (I2PException ie) {
+                    _log.error("Error accepting connection", ie);
+                    return;
+                } catch (ConnectException ce) {
+                    _log.error("Connection already dropped", ce);
+                    return;
+                } catch(SocketTimeoutException ste) {
+                    // ignored
+                }       
+            }
+        }
+        
+        private void handle(I2PSocket sock) {
+            FileOutputStream fos = null;
+            try {
+                File sink = new File(_sinkDir);
+                if (!sink.exists())
+                    sink.mkdirs();
+                File cur = File.createTempFile("clientSink", ".dat", sink);
+                fos = new FileOutputStream(cur);
+                if (_log.shouldLog(Log.DEBUG))
+                    _log.debug("Writing to " + cur.getAbsolutePath());
+            } catch (IOException ioe) {
+                _log.error("Error creating sink", ioe);
+                return;
+            }
+            
+            long start = System.currentTimeMillis();
+            try {
+                InputStream in = sock.getInputStream();
+                byte buf[] = new byte[4096];
+                long written = 0;
+                int read = 0;
+                while ( (read = in.read(buf)) != -1) {
+                    //_fos.write(buf, 0, read);
+                    written += read;
+                    if (_log.shouldLog(Log.DEBUG))
+                        _log.debug("read and wrote " + read + " (" + written + ")");
+                }
+                fos.write(("written: [" + written + "]\n").getBytes());
+                long lifetime = System.currentTimeMillis() - start;
+                _log.info("Got EOF from client socket [written=" + written + " lifetime=" + lifetime + "]");
+            } catch (IOException ioe) {
+                _log.error("Error writing the sink", ioe);
+            } finally {
+                if (fos != null) try { fos.close(); } catch (IOException ioe) {}
+                if (sock != null) try { sock.close(); } catch (IOException ioe) {}
+                _log.debug("Client socket closed");
+            }
+        }
+    }
+    
+    /**
+     * Fire up the streaming server.  <code>Usage: StreamSinkServer [i2cpHost i2cpPort] sinkDir ourDestFile [numHandlers]</code><br>
+     * <ul>
+     *  <li><b>sinkDir</b>: Directory to store received files in</li>
+     *  <li><b>ourDestFile</b>: filename to write our binary destination to</li>
+     *  <li><b>numHandlers</b>: how many concurrent connections to handle</li>
+     * </ul>
+     * @param args [i2cpHost i2cpPort] sinkDir ourDestFile [numHandlers]
+     */
+    public static void main(String args[]) {
+        StreamSinkServer server = null;
+        switch (args.length) {
+            case 0:
+                server = new StreamSinkServer("dataDir", "server.key", "localhost", 7654, 3);
+                break;
+            case 2:
+                server = new StreamSinkServer(args[0], args[1]);
+                break;
+            case 4:
+            case 5:
+                int handlers = 3;
+                if (args.length == 5) {
+                    try {
+                        handlers = Integer.parseInt(args[4]);
+                    } catch (NumberFormatException nfe) {}
+                }
+                try { 
+                    int port = Integer.parseInt(args[1]);
+                    server = new StreamSinkServer(args[2], args[3], args[0], port, handlers);
+                } catch (NumberFormatException nfe) {
+                    System.out.println("Usage: StreamSinkServer [i2cpHost i2cpPort] sinkDir ourDestFile [handlers]");
+                }
+                break;
+            default:
+                System.out.println("Usage: StreamSinkServer [i2cpHost i2cpPort] sinkDir ourDestFile [handlers]");
+        }
+        if (server != null)
+            server.runServer();
+    }
+}

apps/ministreaming/java/demo/net/i2p/client/streaming/StreamSinkTestClient.java

@@ -0,0 +1,33 @@
+package net.i2p.client.streaming;
+
+import net.i2p.client.streaming.StreamSinkSend;
+/**
+ *
+ */
+public class StreamSinkTestClient {
+    public static void main(String args[]) {
+        //System.setProperty(I2PClient.PROP_TCP_HOST, "dev.i2p.net");
+        //System.setProperty(I2PClient.PROP_TCP_PORT, "4501");
+        System.setProperty("tunnels.depthInbound", "0");
+        
+        if (args.length <= 0) {
+            send("/home/jrandom/libjbigi.so");
+        } else {
+            for (int i = 0; i < args.length; i++)
+                send(args[i]);
+        }
+    }
+    
+    private static void send(final String filename) {
+        Thread t = new Thread(new Runnable() { 
+            public void run() { 
+                StreamSinkSend.main(new String[] { filename, "0", "streamSinkTestLiveServer.key" });
+            }
+        }, "client " + filename);
+        t.start();
+        try { t.join(); } catch (Exception e) {}
+        System.err.println("Done sending");
+        try { Thread.sleep(120*1000); } catch (Exception e) {}
+        //System.exit(0);
+    }
+}

apps/ministreaming/java/demo/net/i2p/client/streaming/StreamSinkTestServer.java

@@ -0,0 +1,19 @@
+package net.i2p.client.streaming;
+
+import net.i2p.client.streaming.StreamSinkServer;
+/**
+ *
+ */
+public class StreamSinkTestServer {
+    public static void main(String args[]) {
+        //System.setProperty(I2PClient.PROP_TCP_HOST, "dev.i2p.net");
+        //System.setProperty(I2PClient.PROP_TCP_PORT, "4101");
+        System.setProperty("tunnels.depthInbound", "0");
+        
+        new Thread(new Runnable() { 
+            public void run() { 
+                StreamSinkServer.main(new String[] { "streamSinkTestLiveDir", "streamSinkTestLiveServer.key" });
+            }
+        }, "server").start();
+    }
+}

apps/ministreaming/java/demo/net/i2p/client/streaming/StreamSinkTestStandalone.java

@@ -0,0 +1,93 @@
+package net.i2p.client.streaming;
+
+import net.i2p.client.streaming.StreamSinkClient;
+import net.i2p.client.streaming.StreamSinkSend;
+import net.i2p.client.streaming.StreamSinkServer;
+/**
+ * Usage: StreamSinkTest [(old|new) [#hops [#kb]]]
+ */
+public class StreamSinkTestStandalone {
+/*    private static String HOST1 = "dev.i2p.net";
+    private static String HOST2 = "dev.i2p.net";
+    private static String PORT1 = "4101";
+    private static String PORT2 = "4501";
+ /* 
+    private static String HOST1 = "localhost";
+    private static String HOST2 = "localhost";
+    private static String PORT1 = "7654";
+    private static String PORT2 = "7654";
+   */
+    private static String HOST1 = "localhost";
+    private static String HOST2 = "localhost";
+    private static String PORT1 = "10001";
+    private static String PORT2 = "11001";
+  /* */
+    
+    public static void main(String args[]) {
+        int hops = 0;
+        int kb = 32*1024;
+        if (args.length > 0) {
+            try { 
+                hops = Integer.parseInt(args[0]); 
+            } catch (NumberFormatException nfe) {
+                hops = 0;
+            }
+        }
+        if (args.length > 1) {
+            try {
+                kb = Integer.parseInt(args[1]);
+            } catch (NumberFormatException nfe) {
+                kb = 32*1024;
+            }
+        }
+        
+        System.setProperty("tunnels.depthInbound", ""+hops);
+        
+        new Thread(new Runnable() { 
+            public void run() { 
+                StreamSinkServer.main(new String[] { HOST1, PORT1, "streamSinkTestDir", "streamSinkTestServer.key" });
+            }
+        }, "server").start();
+        
+        try { Thread.sleep(60*1000); } catch (Exception e) {}
+        
+        //run(256, 1);
+        //run(256, 1000);
+        //run(4*1024, 10);
+        run(kb, 1);
+        //run(1*1024, 1);
+        //run("/home/jrandom/streamSinkTestDir/clientSink36766.dat", 1);
+        //run(512*1024, 1);
+        try { Thread.sleep(10*1000); } catch (InterruptedException e) {}
+        System.out.println("Shutting down");
+        System.exit(0);
+    }
+    
+    private static void run(final int kb, final int msBetweenWrites) {
+        Thread t = new Thread(new Runnable() { 
+            public void run() { 
+                StreamSinkClient.main(new String[] { HOST2, PORT2, kb+"", msBetweenWrites+"", "streamSinkTestServer.key" });
+            }
+        });
+        t.start();
+        
+        System.out.println("client and server started: size = " + kb + "KB, delay = " + msBetweenWrites);
+        try {
+            t.join();
+        } catch (InterruptedException ie) {}
+    }
+    
+    private static void run(final String filename, final int msBetweenWrites) {
+        Thread t = new Thread(new Runnable() { 
+            public void run() { 
+                StreamSinkSend.main(new String[] { filename, msBetweenWrites+"", "streamSinkTestServer.key" });
+            }
+        });
+        t.start();
+        
+        System.out.println("client and server started: file " + filename + ", delay = " + msBetweenWrites);
+        try {
+            t.join();
+        } catch (InterruptedException ie) {}
+    }
+}

apps/ministreaming/java/demo/net/i2p/client/streaming/TestSwarm.java

@@ -0,0 +1,216 @@
+package net.i2p.client.streaming;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import net.i2p.I2PAppContext;
+import net.i2p.client.I2PClientFactory;
+import net.i2p.data.Destination;
+import net.i2p.util.I2PThread;
+import net.i2p.util.Log;
+
+/**
+ * Sit around on a destination, receiving lots of data and sending lots of
+ * data to whomever talks to us.
+ *
+ * Usage: TestSwarm myKeyFile [peerDestFile ]*
+ *
+ */
+public class TestSwarm {
+    private I2PAppContext _context;
+    private Log _log;
+    private String _destFile;
+    private String _peerDestFiles[];
+    private I2PSocketManager _manager;
+    private String _conOptions; // unused? used elsewhere?
+    private boolean _dead; // unused? used elsewhere?
+    
+    public static void main(String args[]) {
+        if (args.length < 1) {
+            System.err.println("Usage: TestSwarm myDestFile [peerDestFile ]*");
+            return;
+        }
+        I2PAppContext ctx = new I2PAppContext();
+        String files[] = new String[args.length - 1];
+        System.arraycopy(args, 1, files, 0, files.length);
+        TestSwarm swarm = new TestSwarm(ctx, args[0], files);
+        swarm.startup();
+    }
+    
+    public TestSwarm(I2PAppContext ctx, String destFile, String peerDestFiles[]) {
+        _context = ctx;
+        _log = ctx.logManager().getLog(TestSwarm.class);
+        _dead = false;
+        _destFile = destFile;
+        _peerDestFiles = peerDestFiles;
+        _conOptions = "";
+    }
+    
+    public void startup() {
+        _log.debug("Starting up");
+        File keys = new File(_destFile);
+        if (!keys.exists()) {
+            try {
+                I2PClientFactory.createClient().createDestination(new FileOutputStream(keys));
+            } catch (Exception e) {
+                _log.error("Error creating a new destination on " + keys, e);
+                return;
+            }
+        }
+        try {
+            _manager = I2PSocketManagerFactory.createManager(new FileInputStream(_destFile), null, -1, null);
+        } catch (Exception e) {
+            _log.error("Error creatign the manager", e);
+            return;
+        }
+
+        I2PThread listener = new I2PThread(new Listener(), "Listener");
+        listener.start();
+        
+        connectWithPeers();
+    }
+    
+    
+    private void connectWithPeers() {
+        if (_peerDestFiles != null) {
+            for (int i = 0; i < _peerDestFiles.length; i++) {
+                try {
+                    FileInputStream fin = new FileInputStream(_peerDestFiles[i]);
+                    Destination dest = new Destination();
+                    dest.readBytes(fin);
+                    
+                    I2PThread flooder = new I2PThread(new Flooder(dest), "Flooder+" + dest.calculateHash().toBase64().substring(0,4));
+                    flooder.start();
+                } catch (Exception e) {
+                    _log.error("Unable to read the peer from " + _peerDestFiles[i], e);
+                }
+            }
+        }
+    }
+    
+    private class Listener implements Runnable {
+        public void run() {
+            try {
+                I2PServerSocket ss = _manager.getServerSocket();
+                I2PSocket s = null;
+                while ( (s = ss.accept()) != null) {
+                    I2PThread flooder = new I2PThread(new Flooder(s), "Flooder-" + s.getPeerDestination().calculateHash().toBase64().substring(0,4));
+                    flooder.start();
+                }
+            } catch (Exception e) {
+                _log.error("Error listening", e);
+            }
+        }
+    }
+    
+    private static volatile long __conId = 0;
+    private class Flooder implements Runnable {
+        private Destination _remoteDestination;
+        private I2PSocket _socket;
+        private boolean _closed;
+        private long _started;
+        private long _totalSent;
+        private long _totalReceived;
+        private long _lastReceived;
+        private long _lastReceivedOn;
+        private long _connectionId;
+        
+        public Flooder(Destination dest) {
+            _socket = null;
+            _remoteDestination = dest;
+            _connectionId = ++__conId;
+            _closed = false;
+            _lastReceived = -1;
+            _lastReceivedOn = _context.clock().now();
+            _context.statManager().createRateStat("swarm." + _connectionId + ".totalReceived", "Data size received", "swarm", new long[] { 30*1000, 60*1000, 5*60*1000 });
+            _context.statManager().createRateStat("swarm." + _connectionId + ".totalSent", "Data size sent", "swarm", new long[] { 30*1000, 60*1000, 5*60*1000 });
+            _context.statManager().createRateStat("swarm." + _connectionId + ".started", "When we start", "swarm", new long[] { 5*60*1000 });
+            _context.statManager().createRateStat("swarm." + _connectionId + ".lifetime", "How long we talk to a peer", "swarm", new long[] { 5*60*1000 });
+        }
+        
+        public Flooder(I2PSocket socket) {
+            _socket = socket;
+            _remoteDestination = socket.getPeerDestination();
+            _connectionId = ++__conId;
+            _closed = false;
+            _lastReceived = -1;
+            _lastReceivedOn = _context.clock().now();
+            _context.statManager().createRateStat("swarm." + _connectionId + ".totalReceived", "Data size received", "swarm", new long[] { 30*1000, 60*1000, 5*60*1000 });
+            _context.statManager().createRateStat("swarm." + _connectionId + ".totalSent", "Data size sent", "swarm", new long[] { 30*1000, 60*1000, 5*60*1000 });
+            _context.statManager().createRateStat("swarm." + _connectionId + ".started", "When we start", "swarm", new long[] { 5*60*1000 });
+            _context.statManager().createRateStat("swarm." + _connectionId + ".lifetime", "How long we talk to a peer", "swarm", new long[] { 5*60*1000 });
+        }
+        
+        public long getConnectionId() { return _connectionId; }
+        public Destination getDestination() { return _remoteDestination; }
+        
+        public void run() {
+            _started = _context.clock().now();
+            _context.statManager().addRateData("swarm." + _connectionId + ".started", 1, 0);
+            byte data[] = new byte[4*1024];
+            _context.random().nextBytes(data);
+            long value = 0;
+            long lastSend = _context.clock().now();
+            if (_socket == null) {
+                try {
+                    _socket = _manager.connect(_remoteDestination);
+                } catch (Exception e) {
+                    _log.error("Error connecting to " + _remoteDestination.calculateHash().toBase64().substring(0,4));
+                    return;
+                }
+            }
+            
+            I2PThread floodListener = new I2PThread(new FloodListener(), "FloodListener" + _connectionId);
+            floodListener.start();
+            
+            try {
+                OutputStream out = _socket.getOutputStream();
+                while (!_closed) {
+                    if (shouldSend()) {
+                        out.write(data);
+                        // out.flush();
+                        _totalSent += data.length;
+                        _context.statManager().addRateData("swarm." + _connectionId + ".totalSent", _totalSent, 0);
+                        //try { Thread.sleep(100); } catch (InterruptedException ie) {}
+                        long now = _context.clock().now();
+                        //_log.debug("Sending " + _connectionId + " after " + (now-lastSend));
+                        lastSend = now;
+                        //try { Thread.sleep(20); } catch (InterruptedException ie) {}
+                    } else {
+                        try { Thread.sleep(5000); } catch (InterruptedException ie) {}
+                    }
+                }
+            } catch (Exception e) {
+                _log.error("Error sending", e);
+            }
+        }
+        
+        private class FloodListener implements Runnable {
+            public void run() {
+                long lastRead = System.currentTimeMillis();
+                long now = lastRead;
+                try {
+                    InputStream in = _socket.getInputStream();
+                    byte buf[] = new byte[8*1024];
+                    int read = 0;
+                    while ( (read = in.read(buf)) != -1) {
+                        now = System.currentTimeMillis();
+                        _totalReceived += read;
+                        _context.statManager().addRateData("swarm." + getConnectionId() + ".totalReceived", _totalReceived, 0);
+                        //_log.debug("Receiving " + _connectionId + " with " + read + " after " + (now-lastRead));
+                        lastRead = now;
+                    }
+                } catch (Exception e) {
+                    _log.error("Error listening to the flood", e);
+                }
+            }
+        }
+    }
+    
+    private boolean shouldSend() {
+        return Boolean.valueOf(_context.getProperty("shouldSend", "false")).booleanValue();
+    }
+}