* deal with writes > the packet size limit

* deal with window size > 1, especially before receiving the first ACK
* disable congestion control for the moment (aka unlimited window size)

apps/streaming/java/src/net/i2p/client/streaming/Connection.java

@@ -93,7 +93,7 @@ public class Connection {
      * @return true if the packet should be sent
      */
     boolean packetSendChoke() {
-        //if (true) return true;
+        if (true) return true;
         long writeExpire = _options.getWriteTimeout();
         if (writeExpire > 0)
             writeExpire += _context.clock().now();
@@ -151,30 +151,6 @@ public class Connection {
         _outboundQueue.enqueue(packet);
     }
     
-    /*    
-    void flushPackets() { 
-        List toSend = null;
-        synchronized (_outboundPackets) {
-            for (Iterator iter = _outboundPackets.values().iterator(); iter.hasNext(); ) {
-                PacketLocal packet = (PacketLocal)iter.next();
-                long nextExpected = _options.getResendDelay() << packet.getNumSends();
-                if (packet.getLastSend() + nextExpected <= _context.clock().now()) {
-                    // we need to resend
-                    if (toSend == null) toSend = new ArrayList(1);
-                    toSend.add(packet);
-                }
-            }
-        }
-        
-        if (toSend != null) {
-            for (int i = 0; i < toSend.size(); i++) {
-                PacketLocal packet = (PacketLocal)toSend.get(i);
-                _lastSendTime = _context.clock().now();
-                _outboundQueue.enqueue(packet);
-            }
-        }
-    }
-    */
     List ackPackets(long ackThrough, long nacks[]) {
         List acked = null;
         synchronized (_outboundPackets) {
@@ -362,6 +338,15 @@ public class Connection {
                     resend = true;
             }
             if ( (resend) && (_packet.getAckTime() < 0) ) {
+                // revamp various fields, in case we need to ack more, etc
+                _packet.setAckThrough(getInputStream().getHighestBlockId());
+                _packet.setNacks(getInputStream().getNacks());
+                _packet.setOptionalDelay(getOptions().getChoke());
+                _packet.setOptionalMaxSize(getOptions().getMaxMessageSize());
+                _packet.setResendDelay(getOptions().getResendDelay());
+                _packet.setReceiveStreamId(_receiveStreamId);
+                _packet.setSendStreamId(_sendStreamId);
+
                 if (_log.shouldLog(Log.DEBUG))
                     _log.debug("Resend packet " + _packet + " on " + Connection.this);
                 _outboundQueue.enqueue(_packet);

apps/streaming/java/src/net/i2p/client/streaming/ConnectionPacketHandler.java

@@ -35,16 +35,19 @@ public class ConnectionPacketHandler {
             if (nextTime <= 0) {
                 con.setNextSendTime(con.getOptions().getSendAckDelay() + _context.clock().now());
                 if (_log.shouldLog(Log.DEBUG))
-                    _log.debug("Scheduling ack in " + con.getOptions().getSendAckDelay() + "ms for received packet " + packet);
+                    _log.error("Scheduling ack in " + con.getOptions().getSendAckDelay() + "ms for received packet " + packet);
             } else {
                 if (_log.shouldLog(Log.DEBUG))
-                    _log.debug("Ack is already scheduled in " + nextTime + "ms, though we just received " + packet);
+                    _log.debug("Ack is already scheduled in " + (nextTime-_context.clock().now()) 
+                               + "ms, though we just received " + packet);
             }
         } else {
             if (packet.getSequenceNum() > 0) {
                 // take note of congestion
                 con.getOptions().setResendDelay(con.getOptions().getResendDelay()*2);
                 //con.getOptions().setWindowSize(con.getOptions().getWindowSize()/2);
+                if (_log.shouldLog(Log.WARN))
+                    _log.warn("congestion.. dup " + packet);
             } else {
                 if (_log.shouldLog(Log.DEBUG))
                     _log.debug("ACK only packet received: " + packet);

apps/streaming/java/src/net/i2p/client/streaming/MessageInputStream.java

@@ -95,7 +95,7 @@ public class MessageInputStream extends InputStream {
                 if (_notYetReadyBlocks.containsKey(l)) {
                     // ACK
                 } else {
-                    if (ids != null)
+                    if (ids == null)
                         ids = new ArrayList(4);
                     ids.add(l);
                 }
@@ -210,8 +210,8 @@ public class MessageInputStream extends InputStream {
             expiration = _readTimeout + System.currentTimeMillis();
         synchronized (_dataLock) {
             while (_readyDataBlocks.size() <= 0) {
-                if (_log.shouldLog(Log.DEBUG))
+                //if (_log.shouldLog(Log.DEBUG))
-                    _log.debug("read() with readyBlocks.size = " + _readyDataBlocks.size() + " on " + toString());
+                //    _log.debug("read() with readyBlocks.size = " + _readyDataBlocks.size() + " on " + toString());
                 
                 if ( (_notYetReadyBlocks.size() <= 0) && (_closeReceived) ) {
                     if (_log.shouldLog(Log.DEBUG))
@@ -244,8 +244,8 @@ public class MessageInputStream extends InputStream {
                 }
             }
             
-            if (_log.shouldLog(Log.DEBUG))
+            //if (_log.shouldLog(Log.DEBUG))
-                _log.debug("read() readyBlocks = " + _readyDataBlocks.size() + ": " + toString());
+            //    _log.debug("read() readyBlocks = " + _readyDataBlocks.size() + ": " + toString());
             
             // either was already ready, or we wait()ed and it arrived
             ByteArray cur = (ByteArray)_readyDataBlocks.get(0);

apps/streaming/java/src/net/i2p/client/streaming/MessageOutputStream.java

@@ -20,7 +20,7 @@ public class MessageOutputStream extends OutputStream {
     private boolean _closed;
     
     public MessageOutputStream(I2PAppContext ctx, DataReceiver receiver) {
-        this(ctx, receiver, 64*1024);
+        this(ctx, receiver, Packet.MAX_PAYLOAD_SIZE);
     }
     public MessageOutputStream(I2PAppContext ctx, DataReceiver receiver, int bufSize) {
         super();
@@ -37,31 +37,38 @@ public class MessageOutputStream extends OutputStream {
     }
     
     public void write(byte b[], int off, int len) throws IOException {
+        if (_log.shouldLog(Log.DEBUG))
+            _log.debug("write(b[], " + off + ", " + len + ")");
         synchronized (_dataLock) {
+            int cur = off;
             int remaining = len;
             while (remaining > 0) {
                 if (_valid + remaining < _buf.length) {
                     // simply buffer the data, no flush
-                    System.arraycopy(b, off, _buf, _valid, remaining);
+                    System.arraycopy(b, cur, _buf, _valid, remaining);
                     _valid += remaining;
+                    cur += remaining;
                     remaining = 0;
+                    if (_log.shouldLog(Log.DEBUG))
+                        _log.debug("write(...): appending valid = " + _valid + " remaining=" + remaining);
                 } else {
                     // buffer whatever we can fit then flush,
                     // repeating until we've pushed all of the
                     // data through
                     int toWrite = _buf.length - _valid;
-                    System.arraycopy(b, off, _buf, _valid, toWrite);
+                    System.arraycopy(b, cur, _buf, _valid, toWrite);
                     remaining -= toWrite;
+                    cur += toWrite;
                     _valid = _buf.length;
                     if (_log.shouldLog(Log.DEBUG))
-                        _log.debug("write(b[], " + off + ", " + len + "): valid = " + _valid);
+                        _log.debug("write(...): flushing valid = " + _valid + " remaining=" + remaining);
                     // this blocks until the packet is ack window is open.  it 
                     // also throws InterruptedIOException if the write timeout 
                     // expires
                     _dataReceiver.writeData(_buf, 0, _valid);
                     if (_log.shouldLog(Log.DEBUG))
-                        _log.debug("write(b[], " + off + ", " + len + "): valid = " + _valid + " complete");
+                        _log.debug("write(...): flushing complete valid = " + _valid + " remaining=" + remaining);
-                    _valid = 0;
+                    _valid = 0;                       
                     throwAnyError();
                 }
             }

apps/streaming/java/src/net/i2p/client/streaming/Packet.java

@@ -131,7 +131,7 @@ public class Packet {
      * ping reply (if receiveStreamId is set).
      */
     public static final int FLAG_ECHO = (1 << 9);
-    
+
     public static final int DEFAULT_MAX_SIZE = 32*1024;
     
     /** what stream is this packet a part of? */
@@ -195,9 +195,15 @@ public class Packet {
     public int getResendDelay() { return _resendDelay; }
     public void setResendDelay(int numSeconds) { _resendDelay = numSeconds; }
     
+    public static final int MAX_PAYLOAD_SIZE = 32*1024;
+    
     /** get the actual payload of the message.  may be null */
     public byte[] getPayload() { return _payload; }
-    public void setPayload(byte payload[]) { _payload = payload; }
+    public void setPayload(byte payload[]) { 
+        _payload = payload; 
+        if ( (payload != null) && (payload.length > MAX_PAYLOAD_SIZE) )
+            throw new IllegalArgumentException("Too large payload: " + payload.length);
+    }
 
     /** is a particular flag set on this packet? */
     public boolean isFlagSet(int flag) { return 0 != (_flags & flag); }
@@ -323,7 +329,12 @@ public class Packet {
         }
         
         if (_payload != null) {
-            System.arraycopy(_payload, 0, buffer, cur, _payload.length);
+            try {
+                System.arraycopy(_payload, 0, buffer, cur, _payload.length);
+            } catch (ArrayIndexOutOfBoundsException aioobe) {
+                System.err.println("payload.length: " + _payload.length + " buffer.length: " + buffer.length + " cur: " + cur);
+                throw aioobe;
+            }
             cur += _payload.length;
         }
                 
@@ -411,6 +422,8 @@ public class Packet {
         
         // skip ahead to the payload
         _payload = new byte[offset + length - payloadBegin];
+        if (_payload.length > MAX_PAYLOAD_SIZE)
+            throw new IllegalArgumentException("length: " + length + " offset: " + offset + " begin: " + payloadBegin);
         System.arraycopy(buffer, payloadBegin, _payload, 0, _payload.length);
         
         // ok now lets go back and deal with the options

apps/streaming/java/src/net/i2p/client/streaming/PacketHandler.java

@@ -93,15 +93,28 @@ public class PacketHandler {
                     _log.warn("Syn packet reply on a stream we don't know about: " + packet);
             }
         } else {
+            if (_log.shouldLog(Log.DEBUG))
+                _log.debug("Packet received on an unknown stream (and not a SYN): " + packet);
+            if (packet.getSendStreamId() == null) {
+                for (Iterator iter = _manager.listConnections().iterator(); iter.hasNext(); ) {
+                    Connection con = (Connection)iter.next();
+                    if (DataHelper.eq(con.getSendStreamId(), packet.getReceiveStreamId()) &&
+                        con.getAckedPackets() <= 0) {
+                        if (_log.shouldLog(Log.DEBUG))
+                            _log.debug("Received additional packets before the syn on " + con + ": " + packet);
+                        receiveKnownCon(con, packet);
+                        return;
+                    }
+                }
+            }
             if (_log.shouldLog(Log.WARN)) {
-                _log.warn("Packet received on an unknown stream (and not a SYN): " + packet);
                 StringBuffer buf = new StringBuffer(128);
                 Set cons = _manager.listConnections();
                 for (Iterator iter = cons.iterator(); iter.hasNext(); ) {
                     Connection con = (Connection)iter.next();
                     buf.append(Base64.encode(con.getReceiveStreamId())).append(" ");
                 }
-                _log.warn("Other streams: " + buf.toString());
+                _log.warn("Packet belongs to know other cons: " + packet + " connections: " + buf.toString());
             }
         }
     }

apps/streaming/java/test/net/i2p/client/streaming/EchoLargeTest.java

@@ -0,0 +1,202 @@
+package net.i2p.client.streaming;
+
+import java.io.InputStream;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.OutputStream;
+import java.util.Date;
+import java.util.Properties;
+
+import net.i2p.I2PAppContext;
+import net.i2p.client.I2PClient;
+import net.i2p.client.I2PClientFactory;
+import net.i2p.client.I2PSession;
+import net.i2p.data.Base64;
+import net.i2p.data.DataHelper;
+import net.i2p.data.Destination;
+import net.i2p.util.Log;
+
+/**
+ *
+ */
+public class EchoLargeTest {
+    private Log _log;
+    private I2PSession _client;
+    private I2PSession _server;
+    
+    public void test() {
+        try {
+            I2PAppContext context = I2PAppContext.getGlobalContext();
+            _log = context.logManager().getLog(ConnectTest.class);
+            _log.debug("creating server session");
+            _server = createSession();
+            _log.debug("running server");
+            runServer(context, _server);
+            _log.debug("creating client session");
+            _client = createSession();
+            _log.debug("running client");
+            runClient(context, _client);
+        } catch (Exception e) {
+            _log.error("error running", e);
+        }
+        try { Thread.sleep(300*1000); } catch (Exception e) {}
+    }
+    
+    private void runClient(I2PAppContext ctx, I2PSession session) {
+        Thread t = new Thread(new ClientRunner(ctx, session));
+        t.setName("client");
+        t.setDaemon(true);
+        t.start();
+    }
+    
+    private void runServer(I2PAppContext ctx, I2PSession session) {
+        Thread t = new Thread(new ServerRunner(ctx, session));
+        t.setName("server");
+        t.setDaemon(true);
+        t.start();
+    }
+    
+    private class ServerRunner implements Runnable {
+        private I2PAppContext _context;
+        private I2PSession _session;
+        private Log _log;
+        public ServerRunner(I2PAppContext ctx, I2PSession session) {
+            _context = ctx;
+            _session = session;
+            _log = ctx.logManager().getLog(ServerRunner.class);
+        }
+        
+        public void run() {
+            try {
+                Properties opts = new Properties();
+                I2PSocketManager mgr = new I2PSocketManagerFull(_context, _session, opts, "client");
+                _log.debug("manager created");
+                I2PServerSocket ssocket = mgr.getServerSocket();
+                _log.debug("server socket created");
+                while (true) {
+                    I2PSocket socket = ssocket.accept();
+                    _log.debug("socket accepted: " + socket);
+                    InputStream in = socket.getInputStream();
+                    OutputStream out = socket.getOutputStream();
+                    _log.debug("server streams built");
+                    byte buf[] = new byte[128*1024];
+                    while (buf != null) {
+                        for (int i = 0; i < buf.length; i++) {
+                            int c = in.read();
+                            if (c == -1) {
+                                buf = null;
+                                break;
+                            } else {
+                                buf[i] = (byte)(c & 0xFF);
+                            }
+                        }
+                        if (buf != null) {
+                            _log.debug("* server read the full buffer");
+                            out.write(buf);
+                            out.flush();
+                        }
+                    }
+                    if (_log.shouldLog(Log.DEBUG))
+                        _log.debug("Closing the received server socket");
+                    socket.close();
+                }
+            } catch (Exception e) {
+                _log.error("error running", e);
+            }
+        }
+        
+    }
+    
+    private class ClientRunner implements Runnable {
+        private I2PAppContext _context;
+        private I2PSession _session;
+        private Log _log;
+        public ClientRunner(I2PAppContext ctx, I2PSession session) {
+            _context = ctx;
+            _session = session;
+            _log = ctx.logManager().getLog(ClientRunner.class);
+        }
+        
+        public void run() {
+            try {
+                Properties opts = new Properties();
+                I2PSocketManager mgr = new I2PSocketManagerFull(_context, _session, opts, "client");
+                _log.debug("manager created");
+                I2PSocket socket = mgr.connect(_server.getMyDestination());
+                _log.debug("socket created");
+                InputStream in = socket.getInputStream();
+                OutputStream out = socket.getOutputStream();
+                for (int i = 0; i < 3; i++) {
+                    byte buf[] = new byte[128*1024];
+                    _context.random().nextBytes(buf);
+                    byte orig[] = new byte[buf.length];
+                    System.arraycopy(buf, 0, orig, 0, buf.length);
+                    out.write(buf);
+                    _log.debug("client wrote a buffer");
+                    out.flush();
+                    _log.debug("client flushed");
+                    
+                    byte rbuf[] = new byte[buf.length];
+                    for (int j = 0; j < buf.length; j++) {
+                        int c = in.read();
+                        if (c == -1) {
+                            buf = null;
+                            break;
+                        } else {                
+                            //_log.debug("client read: " + ((char)c));
+                            if (c < 0) c += 256;
+                            rbuf[j] = (byte)(c & 0xFF);
+                        }
+                    }
+                    if (buf != null) {
+                        _log.debug("* client read a full buffer");
+                        int firstOff = -1;
+                        for (int k = 0; k < orig.length; k++) {
+                            if (orig[k] != rbuf[k]) {
+                                firstOff = k;
+                                break;
+                            }
+                        }
+                        if (firstOff < 0) {
+                            System.out.println("** Read match");
+                        } else {
+                            System.out.println("** Read does not match: first off = " + firstOff);
+                            _log.error("read does not match (first off = " + firstOff + "): \n"
+                                        + Base64.encode(orig) + "\n" 
+                                        + Base64.encode(rbuf));
+                        }
+                    }
+                }
+                if (_log.shouldLog(Log.DEBUG))
+                    _log.debug("Closing the client socket");
+                socket.close();
+                _log.debug("socket closed");
+                
+                Thread.sleep(5*1000);
+                System.exit(0);
+            } catch (Exception e) {
+                _log.error("error running", e);
+            }
+        }
+        
+    }
+    
+    private I2PSession createSession() {
+        try {
+            I2PClient client = I2PClientFactory.createClient();
+            ByteArrayOutputStream baos = new ByteArrayOutputStream(512);
+            Destination dest = client.createDestination(baos);
+            I2PSession sess = client.createSession(new ByteArrayInputStream(baos.toByteArray()), new Properties());
+            sess.connect();
+            return sess;
+        } catch (Exception e) {
+            _log.error("error running", e);
+            throw new RuntimeException("b0rk b0rk b0rk");
+        }
+    }
+    
+    public static void main(String args[]) {
+        EchoLargeTest et = new EchoLargeTest();
+        et.test();
+    }
+}

apps/streaming/java/test/net/i2p/client/streaming/MessageInputStreamTest.java

@@ -22,17 +22,17 @@ public class MessageInputStreamTest {
     }
     
     public void testInOrder() {
-        byte orig[] = new byte[32*1024];
+        byte orig[] = new byte[256*1024];
         _context.random().nextBytes(orig);
         
         MessageInputStream in = new MessageInputStream(_context);
-        for (int i = 0; i < 32; i++) {
+        for (int i = 0; i < orig.length / 1024; i++) {
             byte msg[] = new byte[1024];
             System.arraycopy(orig, i*1024, msg, 0, 1024);
             in.messageReceived(i, msg);
         }
         
-        byte read[] = new byte[32*1024];
+        byte read[] = new byte[orig.length];
         try {
             int howMany = DataHelper.read(in, read);
             if (howMany != orig.length)
@@ -47,15 +47,15 @@ public class MessageInputStreamTest {
     }
     
     public void testRandomOrder() {
-        byte orig[] = new byte[32*1024];
+        byte orig[] = new byte[256*1024];
         _context.random().nextBytes(orig);
         
         MessageInputStream in = new MessageInputStream(_context);
         ArrayList order = new ArrayList(32);
-        for (int i = 0; i < 32; i++)
+        for (int i = 0; i < orig.length / 1024; i++)
             order.add(new Integer(i));
         Collections.shuffle(order);
-        for (int i = 0; i < 32; i++) {
+        for (int i = 0; i < orig.length / 1024; i++) {
             byte msg[] = new byte[1024];
             Integer cur = (Integer)order.get(i);
             System.arraycopy(orig, cur.intValue()*1024, msg, 0, 1024);
@@ -63,7 +63,7 @@ public class MessageInputStreamTest {
             _log.debug("Injecting " + cur);
         }
         
-        byte read[] = new byte[32*1024];
+        byte read[] = new byte[orig.length];
         try {
             int howMany = DataHelper.read(in, read);
             if (howMany != orig.length)
@@ -77,11 +77,45 @@ public class MessageInputStreamTest {
         }
     }
     
+    public void testRandomDups() {
+        byte orig[] = new byte[256*1024];
+        _context.random().nextBytes(orig);
+        
+        MessageInputStream in = new MessageInputStream(_context);
+        for (int n = 0; n < 3; n++) {
+            ArrayList order = new ArrayList(32);
+            for (int i = 0; i < orig.length / 1024; i++)
+                order.add(new Integer(i));
+            Collections.shuffle(order);
+            for (int i = 0; i < orig.length / 1024; i++) {
+                byte msg[] = new byte[1024];
+                Integer cur = (Integer)order.get(i);
+                System.arraycopy(orig, cur.intValue()*1024, msg, 0, 1024);
+                in.messageReceived(cur.intValue(), msg);
+                _log.debug("Injecting " + cur);
+            }
+        }
+        
+        byte read[] = new byte[orig.length];
+        try {
+            int howMany = DataHelper.read(in, read);
+            if (howMany != orig.length)
+                throw new RuntimeException("Failed test: not enough bytes read [" + howMany + "]");
+            if (!DataHelper.eq(orig, read))
+                throw new RuntimeException("Failed test: data read is not equal");
+            
+            _log.info("Passed test: random dups");
+        } catch (IOException ioe) {
+            throw new RuntimeException("IOError reading: " + ioe.getMessage());
+        }
+    }
+    
     public static void main(String args[]) {
         MessageInputStreamTest t = new MessageInputStreamTest();
         try {
             t.testInOrder();
             t.testRandomOrder();
+            t.testRandomDups();
         } catch (Exception e) {
             e.printStackTrace();
         }

apps/streaming/java/test/net/i2p/client/streaming/MessageOutputStreamTest.java

@@ -0,0 +1,66 @@
+package net.i2p.client.streaming;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+
+import net.i2p.I2PAppContext;
+import net.i2p.data.Base64;
+import net.i2p.data.DataHelper;
+import net.i2p.util.Log;
+
+/**
+ *
+ */
+public class MessageOutputStreamTest {
+    private I2PAppContext _context;
+    private Log _log;
+    
+    public MessageOutputStreamTest() {
+        _context = I2PAppContext.getGlobalContext();
+        _log = _context.logManager().getLog(MessageOutputStreamTest.class);
+    }
+    
+    public void test() {
+        Receiver receiver = new Receiver();
+        MessageOutputStream out = new MessageOutputStream(_context, receiver);
+        byte buf[] = new byte[128*1024];
+        _context.random().nextBytes(buf);
+        try {
+            out.write(buf);
+            out.flush();
+        } catch (IOException ioe) { ioe.printStackTrace(); }
+        byte read[] = receiver.getData();
+        int firstOff = -1;
+        for (int k = 0; k < buf.length; k++) {
+            if (buf[k] != read[k]) {
+                firstOff = k;
+                break;
+            }
+        }
+        if (firstOff < 0) {
+            System.out.println("** Read match");
+        } else {
+            System.out.println("** Read does not match: first off = " + firstOff);
+            _log.error("read does not match (first off = " + firstOff + "): \n"
+                        + Base64.encode(buf) + "\n" 
+                        + Base64.encode(read));
+        }
+    }
+    
+    private class Receiver implements MessageOutputStream.DataReceiver {
+        private ByteArrayOutputStream _data;
+        public Receiver() {
+            _data = new ByteArrayOutputStream();
+        }
+        public void writeData(byte[] buf, int off, int size) throws IOException {
+            _data.write(buf, off, size);
+        }
+        public byte[] getData() { return _data.toByteArray(); }
+    }
+    
+    public static void main(String args[]) {
+        MessageOutputStreamTest t = new MessageOutputStreamTest();
+        t.test();
+    }
+}