very basic tests pass (ping, open then pause then close, open then echo back and forth a few times then close)

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

@@ -0,0 +1,386 @@
+package net.i2p.client.streaming;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import net.i2p.I2PAppContext;
+import net.i2p.client.I2PSession;
+import net.i2p.data.Base64;
+import net.i2p.data.Destination;
+import net.i2p.util.Log;
+import net.i2p.util.SimpleTimer;
+
+/**
+ * Maintain the state controlling a streaming connection between two 
+ * destinations.
+ *
+ */
+public class Connection {
+    private I2PAppContext _context;
+    private Log _log;
+    private ConnectionManager _connectionManager;
+    private Destination _remotePeer;
+    private byte _sendStreamId[];
+    private byte _receiveStreamId[];
+    private long _lastSendTime;
+    private long _lastSendId;
+    private boolean _resetReceived;
+    private boolean _connected;
+    private MessageInputStream _inputStream;
+    private MessageOutputStream _outputStream;
+    private SchedulerChooser _chooser;
+    private long _nextSendTime;
+    private long _ackedPackets;
+    private long _createdOn;
+    private long _closeSentOn;
+    private long _closeReceivedOn;
+    private int _unackedPacketsReceived;
+    /** Packet ID (Long) to PacketLocal for sent but unacked packets */
+    private Map _outboundPackets;
+    private PacketQueue _outboundQueue;
+    private ConnectionPacketHandler _handler;
+    private ConnectionOptions _options;
+    private ConnectionDataReceiver _receiver;
+    private I2PSocketFull _socket;
+    /** set to an error cause if the connection could not be established */
+    private String _connectionError;
+    
+    public Connection(I2PAppContext ctx, ConnectionManager manager, SchedulerChooser chooser, PacketQueue queue, ConnectionPacketHandler handler) {
+        this(ctx, manager, chooser, queue, handler, null);
+    }
+    public Connection(I2PAppContext ctx, ConnectionManager manager, SchedulerChooser chooser, PacketQueue queue, ConnectionPacketHandler handler, ConnectionOptions opts) {
+        _context = ctx;
+        _log = ctx.logManager().getLog(Connection.class);
+        _receiver = new ConnectionDataReceiver(ctx, this);
+        _inputStream = new MessageInputStream(ctx);
+        _outputStream = new MessageOutputStream(ctx, _receiver);
+        _chooser = chooser;
+        _outboundPackets = new TreeMap();
+        _outboundQueue = queue;
+        _handler = handler;
+        _options = (opts != null ? opts : new ConnectionOptions());
+        _lastSendId = -1;
+        _nextSendTime = -1;
+        _ackedPackets = 0;
+        _createdOn = ctx.clock().now();
+        _closeSentOn = -1;
+        _closeReceivedOn = -1;
+        _unackedPacketsReceived = 0;
+        _connectionManager = manager;
+        _resetReceived = false;
+        _connected = true;
+    }
+    
+    public long getNextOutboundPacketNum() { 
+        synchronized (this) {
+            return ++_lastSendId;
+        }
+    }
+    
+    void closeReceived() {
+        setCloseReceivedOn(_context.clock().now());
+        _inputStream.closeReceived();
+    }
+    
+    /**
+     * Block until there is an open outbound packet slot or the write timeout 
+     * expires.  
+     *
+     * @return true if the packet should be sent
+     */
+    boolean packetSendChoke() {
+        //if (true) return true;
+        long writeExpire = _options.getWriteTimeout();
+        if (writeExpire > 0)
+            writeExpire += _context.clock().now();
+        while (true) {
+            long timeLeft = writeExpire - _context.clock().now();
+            synchronized (_outboundPackets) {
+                if (_outboundPackets.size() >= _options.getWindowSize()) {
+                    if (writeExpire > 0) {
+                        if (timeLeft <= 0) return false;
+                        if (_log.shouldLog(Log.DEBUG))
+                            _log.debug("Outbound window is full (" + _outboundPackets.size() + "/" + _options.getWindowSize() + "), waiting " + timeLeft);
+                        try { _outboundPackets.wait(timeLeft); } catch (InterruptedException ie) {}
+                    } else {
+                        if (_log.shouldLog(Log.DEBUG))
+                            _log.debug("Outbound window is full (" + _outboundPackets.size() + "), waiting indefinitely");
+                        try { _outboundPackets.wait(); } catch (InterruptedException ie) {}
+                    }
+                } else {
+                    return true;
+                }
+            }
+        }
+    }
+    
+    /**
+     * Flush any data that we can
+     */
+    void sendAvailable() {
+        // this grabs the data, builds a packet, and queues it up via sendPacket
+        try {
+            _outputStream.flushAvailable(_receiver);
+        } catch (IOException ioe) {
+            if (_log.shouldLog(Log.ERROR))
+                _log.error("Error flushing available", ioe);
+        }
+    }
+    
+    void sendPacket(PacketLocal packet) {
+        setNextSendTime(-1);
+        _unackedPacketsReceived = 0;
+        if (_options.getRequireFullySigned()) {
+            packet.setFlag(Packet.FLAG_SIGNATURE_INCLUDED);
+            packet.setFlag(Packet.FLAG_SIGNATURE_REQUESTED);
+        }
+        
+        if ( (packet.getSequenceNum() == 0) && (!packet.isFlagSet(Packet.FLAG_SYNCHRONIZE)) ) {
+            // ACK only, no retries
+        } else {
+            synchronized (_outboundPackets) {
+                _outboundPackets.put(new Long(packet.getSequenceNum()), packet);
+            }
+            SimpleTimer.getInstance().addEvent(new ResendPacketEvent(packet), _options.getResendDelay());
+        }
+        _lastSendTime = _context.clock().now();
+        _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) {
+            for (Iterator iter = _outboundPackets.keySet().iterator(); iter.hasNext(); ) {
+                Long id = (Long)iter.next();
+                if (id.longValue() <= ackThrough) {
+                    if (nacks != null) {
+                        // linear search since its probably really tiny
+                        for (int i = 0; i < nacks.length; i++)
+                            if (nacks[i] == id.longValue())
+                                continue; // NACKed
+                    } else {
+                        // ACKed
+                        if (acked == null) 
+                            acked = new ArrayList(1);
+                        PacketLocal ackedPacket = (PacketLocal)_outboundPackets.get(id);
+                        ackedPacket.ackReceived();
+                        acked.add(ackedPacket);
+                    }
+                } else {
+                    break; // _outboundPackets is ordered
+                }
+            }
+            if (acked != null) {
+                for (int i = 0; i < acked.size(); i++) {
+                    PacketLocal p = (PacketLocal)acked.get(i);
+                    _outboundPackets.remove(new Long(p.getSequenceNum()));
+                    _ackedPackets++;
+                }
+            }
+            _outboundPackets.notifyAll();
+        }
+        return acked;
+    }
+
+    void eventOccurred() {
+        _chooser.getScheduler(this).eventOccurred(this);
+    }
+    
+    void resetReceived() {
+        _resetReceived = true;
+        _outputStream.streamErrorOccurred(new IOException("Reset received"));
+        _inputStream.streamErrorOccurred(new IOException("Reset received"));
+    }
+    public boolean getResetReceived() { return _resetReceived; }
+    
+    public boolean getIsConnected() { return _connected; }
+
+    void disconnect(boolean cleanDisconnect) {
+        if (!_connected) return;
+        _connected = false;
+        if (_log.shouldLog(Log.DEBUG))
+            _log.debug("Disconnecting " + toString(), new Exception("discon"));
+        
+        if (cleanDisconnect) {
+            // send close packets and schedule stuff...
+            try { 
+                _outputStream.close();
+                _inputStream.close();
+            } catch (IOException ioe) {
+                if (_log.shouldLog(Log.WARN))
+                    _log.warn("Error on clean disconnect", ioe);
+            }
+        } else {
+            doClose();
+            synchronized (_outboundPackets) {
+                _outboundPackets.clear();
+            }
+            _connectionManager.removeConnection(this);
+        }
+    }
+    
+    void disconnectComplete() {
+        _connectionManager.removeConnection(this);
+    }
+    
+    private void doClose() {
+        _outputStream.streamErrorOccurred(new IOException("Hard disconnect"));
+        _inputStream.closeReceived();
+    }
+    
+    /** who are we talking with */
+    public Destination getRemotePeer() { return _remotePeer; }
+    public void setRemotePeer(Destination peer) { _remotePeer = peer; }
+    
+    /** what stream do we send data to the peer on? */
+    public byte[] getSendStreamId() { return _sendStreamId; }
+    public void setSendStreamId(byte[] id) { _sendStreamId = id; }
+    
+    /** what stream does the peer send data to us on? (may be null) */
+    public byte[] getReceiveStreamId() { return _receiveStreamId; }
+    public void setReceiveStreamId(byte[] id) { _receiveStreamId = id; }
+    
+    /** when did we last send anything to the peer? */
+    public long getLastSendTime() { return _lastSendTime; }
+    public void setLastSendTime(long when) { _lastSendTime = when; }
+    
+    /** what was the last packet Id sent to the peer? */
+    public long getLastSendId() { return _lastSendId; }
+    public void setLastSendId(long id) { _lastSendId = id; }
+    
+    public ConnectionOptions getOptions() { return _options; }
+    public void setOptions(ConnectionOptions opts) { _options = opts; }
+        
+    public I2PSession getSession() { return _connectionManager.getSession(); }
+    public I2PSocketFull getSocket() { return _socket; }
+    public void setSocket(I2PSocketFull socket) { _socket = socket; }
+    
+    public String getConnectionError() { return _connectionError; }
+    public void setConnectionError(String err) { _connectionError = err; }
+    
+    public ConnectionPacketHandler getPacketHandler() { return _handler; }
+    
+    /** 
+     * when does the scheduler next want to send a packet?   -1 if never.
+     * This should be set when we want to send on timeout, for instance, or
+     * want to delay an ACK.
+     */
+    public long getNextSendTime() { return _nextSendTime; }
+    public void setNextSendTime(long when) { 
+        if (_nextSendTime > 0)
+            if (_log.shouldLog(Log.DEBUG))
+                _log.debug("set next send time to " + (when-_nextSendTime) + "ms after it was before ("+when+")");
+        _nextSendTime = when; 
+    }
+    
+    public long getAckedPackets() { return _ackedPackets; }
+    public long getCreatedOn() { return _createdOn; }
+    public long getCloseSentOn() { return _closeSentOn; }
+    public void setCloseSentOn(long when) { _closeSentOn = when; }
+    public long getCloseReceivedOn() { return _closeReceivedOn; }
+    public void setCloseReceivedOn(long when) { _closeReceivedOn = when; }
+    
+    public void incrementUnackedPacketsReceived() { _unackedPacketsReceived++; }
+    public int getUnackedPacketsReceived() { return _unackedPacketsReceived; }
+    public int getUnackedPacketsSent() { 
+        synchronized (_outboundPackets) { 
+            return _outboundPackets.size(); 
+        } 
+    }
+    
+    /** stream that the local peer receives data on */
+    public MessageInputStream getInputStream() { return _inputStream; }
+    /** stream that the local peer sends data to the remote peer on */
+    public MessageOutputStream getOutputStream() { return _outputStream; }
+    
+    public String toString() { 
+        StringBuffer buf = new StringBuffer(128);
+        buf.append("[Connection ");
+        if (_receiveStreamId != null)
+            buf.append(Base64.encode(_receiveStreamId));
+        else
+            buf.append("unknown");
+        buf.append("<-->");
+        if (_sendStreamId != null)
+            buf.append(Base64.encode(_sendStreamId));
+        else
+            buf.append("unknown");
+        buf.append(" unacked outbound: ");
+        synchronized (_outboundPackets) {
+            buf.append(_outboundPackets.size()).append(" [");
+            for (Iterator iter = _outboundPackets.keySet().iterator(); iter.hasNext(); ) {
+                buf.append(((Long)iter.next()).longValue()).append(" ");
+            }
+            buf.append("] ");
+        }
+        buf.append("unacked inbound? ").append(getUnackedPacketsReceived());
+        buf.append("]");
+        return buf.toString();
+    }
+    
+    /**
+     * Coordinate the resends of a given packet
+     */
+    private class ResendPacketEvent implements SimpleTimer.TimedEvent {
+        private PacketLocal _packet;
+        public ResendPacketEvent(PacketLocal packet) {
+            _packet = packet;
+        }
+        
+        public void timeReached() {
+            boolean resend = false;
+            synchronized (_outboundPackets) {
+                if (_outboundPackets.containsKey(new Long(_packet.getSequenceNum())))
+                    resend = true;
+            }
+            if ( (resend) && (_packet.getAckTime() < 0) ) {
+                if (_log.shouldLog(Log.DEBUG))
+                    _log.debug("Resend packet " + _packet + " on " + Connection.this);
+                _outboundQueue.enqueue(_packet);
+                
+                int numSends = _packet.getNumSends();
+                if (numSends > _options.getMaxResends()) {
+                    if (_log.shouldLog(Log.DEBUG))
+                        _log.debug("Too many resends");
+                    disconnect(false);
+                } else {
+                    long timeout = _options.getResendDelay() << numSends;
+                    if (_log.shouldLog(Log.DEBUG))
+                        _log.debug("Scheduling resend in " + timeout + "ms");
+                    SimpleTimer.getInstance().addEvent(ResendPacketEvent.this, timeout);
+                }
+            } else {
+                if (_log.shouldLog(Log.DEBUG))
+                    _log.debug("Packet acked before resend: " + _packet + " on " + Connection.this);
+            }
+        }
+    }
+}

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

@@ -0,0 +1,104 @@
+package net.i2p.client.streaming;
+
+import java.io.InterruptedIOException;
+import net.i2p.I2PAppContext;
+import net.i2p.util.Log;
+
+/**
+ *
+ */
+class ConnectionDataReceiver implements MessageOutputStream.DataReceiver {
+    private I2PAppContext _context;
+    private Log _log;
+    private Connection _connection;
+    
+    public ConnectionDataReceiver(I2PAppContext ctx, Connection con) {
+        _context = ctx;
+        _log = ctx.logManager().getLog(ConnectionDataReceiver.class);
+        _connection = con;
+    }
+    
+    public void writeData(byte[] buf, int off, int size) throws InterruptedIOException {
+        if (!_connection.packetSendChoke())
+            throw new InterruptedIOException("Timeout expired waiting to write");
+        boolean doSend = true;
+        if ( (size <= 0) && (_connection.getLastSendId() >= 0) ) {
+            if (_connection.getOutputStream().getClosed()) {
+                if (_connection.getCloseSentOn() < 0) {
+                    doSend = true;
+                } else {
+                    // closed, no new data, and we've already sent a close packet
+                    doSend = false;
+                }
+            } else {
+                // no new data, not closed, already synchronized
+                doSend = false;
+            }
+        }
+        
+        if (_connection.getUnackedPacketsReceived() > 0)
+            doSend = true;
+        
+        if (_log.shouldLog(Log.DEBUG))
+            _log.debug("writeData called: size="+size + " doSend=" + doSend + " con: " + _connection, new Exception("write called by"));
+
+        if (doSend) {
+            PacketLocal packet = buildPacket(buf, off, size);
+            _connection.sendPacket(packet);
+        } else {
+            //_connection.flushPackets();
+        }
+    }
+    
+    private boolean isAckOnly(int size) {
+        return ( (size <= 0) && // no data
+                 (_connection.getLastSendId() >= 0) && // not a SYN
+                 ( (!_connection.getOutputStream().getClosed()) || // not a CLOSE
+                 (_connection.getOutputStream().getClosed() && 
+                  _connection.getCloseSentOn() > 0) )); // or it is a dup CLOSE
+    }
+    
+    private PacketLocal buildPacket(byte buf[], int off, int size) {
+        boolean ackOnly = isAckOnly(size);
+        PacketLocal packet = new PacketLocal(_context, _connection.getRemotePeer());
+        byte data[] = new byte[size];
+        System.arraycopy(buf, off, data, 0, size);
+        packet.setPayload(data);
+		if (ackOnly)
+			packet.setSequenceNum(0);
+        else
+            packet.setSequenceNum(_connection.getNextOutboundPacketNum());
+        packet.setSendStreamId(_connection.getSendStreamId());
+        packet.setReceiveStreamId(_connection.getReceiveStreamId());
+        
+        packet.setAckThrough(_connection.getInputStream().getHighestBlockId());
+        packet.setNacks(_connection.getInputStream().getNacks());
+        packet.setOptionalDelay(_connection.getOptions().getChoke());
+        packet.setOptionalMaxSize(_connection.getOptions().getMaxMessageSize());
+        packet.setResendDelay(_connection.getOptions().getResendDelay());
+        
+        if (_connection.getOptions().getProfile() == ConnectionOptions.PROFILE_INTERACTIVE)
+            packet.setFlag(Packet.FLAG_PROFILE_INTERACTIVE, true);
+        else
+            packet.setFlag(Packet.FLAG_PROFILE_INTERACTIVE, false);
+        
+        packet.setFlag(Packet.FLAG_SIGNATURE_REQUESTED, _connection.getOptions().getRequireFullySigned());
+        
+        if ( (!ackOnly) && (packet.getSequenceNum() <= 0) ) {
+            packet.setFlag(Packet.FLAG_SYNCHRONIZE);
+            packet.setOptionalFrom(_connection.getSession().getMyDestination());
+        }
+        
+        if (_connection.getOutputStream().getClosed()) {
+            packet.setFlag(Packet.FLAG_CLOSE);
+            _connection.setCloseSentOn(_context.clock().now());
+            if (_log.shouldLog(Log.DEBUG))
+                _log.debug("Closed is set for a new packet on " + _connection + ": " + packet);
+        } else {
+            if (_log.shouldLog(Log.DEBUG))
+                _log.debug("Closed is not set for a new packet on " + _connection + ": " + packet);
+        }
+        return packet;
+    }
+    
+}

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

@@ -0,0 +1,126 @@
+package net.i2p.client.streaming;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import net.i2p.I2PAppContext;
+import net.i2p.util.Log;
+import net.i2p.util.SimpleTimer;
+
+/**
+ * Receive new connection attempts
+ */
+class ConnectionHandler {
+    private I2PAppContext _context;
+    private Log _log;
+    private ConnectionManager _manager;
+    private List _synQueue;
+    private boolean _active;
+    private int _acceptTimeout;
+    
+    /** max time after receiveNewSyn() and before the matched accept() */
+    private static final int DEFAULT_ACCEPT_TIMEOUT = 3*1000;
+    
+    /** Creates a new instance of ConnectionHandler */
+    public ConnectionHandler(I2PAppContext context, ConnectionManager mgr) {
+        _context = context;
+        _log = context.logManager().getLog(ConnectionHandler.class);
+        _manager = mgr;
+        _synQueue = new ArrayList(5);
+        _active = false;
+        _acceptTimeout = DEFAULT_ACCEPT_TIMEOUT;
+    }
+    
+    public void setActive(boolean active) { _active = active; }
+    public boolean getActive() { return _active; }
+    
+    public void receiveNewSyn(Packet packet) {
+        if (!_active) {
+            if (_log.shouldLog(Log.WARN))
+                _log.warn("Dropping new SYN request, as we're not listening");
+            sendReset(packet);
+            return;
+        }
+        if (_log.shouldLog(Log.DEBUG))
+            _log.debug("Receive new SYN: " + packet + ": timeout in " + _acceptTimeout);
+        SimpleTimer.getInstance().addEvent(new TimeoutSyn(packet), _acceptTimeout);
+        synchronized (_synQueue) {
+            _synQueue.add(packet);
+            _synQueue.notifyAll();
+        }
+    }
+    
+    public Connection accept(long timeoutMs) {
+        if (_log.shouldLog(Log.DEBUG))
+            _log.debug("Accept("+ timeoutMs+") called");
+
+        long expiration = timeoutMs;
+        if (expiration > 0)
+            expiration += _context.clock().now();
+        Packet syn = null;
+        synchronized (_synQueue) {
+            while ( _active && (_synQueue.size() <= 0) ) {
+                if (_log.shouldLog(Log.DEBUG))
+                    _log.debug("Accept("+ timeoutMs+"): active=" + _active + " queue: " + _synQueue.size());
+                if (timeoutMs <= 0) {
+                    try { _synQueue.wait(); } catch (InterruptedException ie) {}
+                } else {
+                    long remaining = expiration - _context.clock().now();
+                    if (remaining < 0)
+                        break;
+                    try { _synQueue.wait(remaining); } catch (InterruptedException ie) {}
+                }
+            }
+            if (_active && _synQueue.size() > 0) {
+                syn = (Packet)_synQueue.remove(0);
+            }
+        }
+        
+        if (syn != null) {
+			return _manager.receiveConnection(syn);
+        } else {
+            return null;
+        }
+    }
+    
+    private void sendReset(Packet packet) {
+        boolean ok = packet.verifySignature(_context, packet.getOptionalFrom(), null);
+        if (!ok) {
+            if (_log.shouldLog(Log.WARN))
+                _log.warn("Received a spoofed SYN packet: they said they were " + packet.getOptionalFrom());
+            return;
+        }
+        PacketLocal reply = new PacketLocal(_context, packet.getOptionalFrom());
+        reply.setFlag(Packet.FLAG_RESET);
+        reply.setFlag(Packet.FLAG_SIGNATURE_INCLUDED);
+        reply.setAckThrough(packet.getSequenceNum());
+        reply.setSendStreamId(packet.getReceiveStreamId());
+        reply.setReceiveStreamId(null);
+        reply.setOptionalFrom(_manager.getSession().getMyDestination());
+        if (_log.shouldLog(Log.DEBUG))
+            _log.debug("Sending RST: " + reply + " because of " + packet);
+        // this just sends the packet - no retries or whatnot
+        _manager.getPacketQueue().enqueue(reply);
+    }
+    
+    private class TimeoutSyn implements SimpleTimer.TimedEvent {
+        private Packet _synPacket;
+        public TimeoutSyn(Packet packet) {
+            _synPacket = packet;
+        }
+        
+        public void timeReached() {
+            boolean removed = false;
+            synchronized (_synQueue) {
+                removed = _synQueue.remove(_synPacket);
+            }
+            
+            if (removed) {
+                // timeout - send RST
+                sendReset(_synPacket);
+            } else {
+                // handled.  noop
+            }
+        }
+    }
+}

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

@@ -0,0 +1,200 @@
+package net.i2p.client.streaming;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
+
+import net.i2p.I2PAppContext;
+import net.i2p.client.I2PSession;
+import net.i2p.data.ByteArray;
+import net.i2p.data.Destination;
+
+/**
+ * Coordinate all of the connections for a single local destination.
+ *
+ *
+ */
+public class ConnectionManager {
+    private I2PAppContext _context;
+    private I2PSession _session;
+    private MessageHandler _messageHandler;
+    private PacketHandler _packetHandler;
+    private ConnectionHandler _connectionHandler;
+    private PacketQueue _outboundQueue;
+    private SchedulerChooser _schedulerChooser;
+    private ConnectionPacketHandler _conPacketHandler;
+    /** Inbound stream ID (ByteArray) to Connection map */
+    private Map _connectionByInboundId;
+    /** Ping ID (ByteArray) to PingRequest */
+    private Map _pendingPings;
+    private boolean _allowIncoming;
+    private Object _connectionLock;
+    
+    public ConnectionManager(I2PAppContext context, I2PSession session) {
+        _context = context;
+        _connectionByInboundId = new HashMap(32);
+        _pendingPings = new HashMap(4);
+        _connectionLock = new Object();
+        _messageHandler = new MessageHandler(context, this);
+        _packetHandler = new PacketHandler(context, this);
+        _connectionHandler = new ConnectionHandler(context, this);
+        _schedulerChooser = new SchedulerChooser(context);
+        _conPacketHandler = new ConnectionPacketHandler(context);
+        _session = session;
+        session.setSessionListener(_messageHandler);
+        _outboundQueue = new PacketQueue(context, session);
+        _allowIncoming = false;
+    }
+    
+    Connection getConnectionByInboundId(byte[] id) {
+        synchronized (_connectionLock) {
+            return (Connection)_connectionByInboundId.get(new ByteArray(id));
+        }
+    }
+    
+    public void setAllowIncomingConnections(boolean allow) { 
+        _connectionHandler.setActive(allow);
+    }
+    public boolean getAllowIncomingConnections() {
+        return _connectionHandler.getActive();
+    }
+    
+    /**
+     * Create a new connection based on the SYN packet we received.
+     *
+     * @return created Connection with the packet's data already delivered to
+     *         it, or null if the syn's streamId was already taken
+     */
+    public Connection receiveConnection(Packet synPacket) {
+        Connection con = new Connection(_context, this, _schedulerChooser, _outboundQueue, _conPacketHandler);
+        byte receiveId[] = new byte[4];
+        _context.random().nextBytes(receiveId);
+        synchronized (_connectionLock) {
+            while (true) {
+                Connection oldCon = (Connection)_connectionByInboundId.put(new ByteArray(receiveId), con);
+                if (oldCon == null) {
+                    break;
+                } else { 
+                    _connectionByInboundId.put(new ByteArray(receiveId), oldCon);
+                    // receiveId already taken, try another
+                    _context.random().nextBytes(receiveId);
+                }
+            }
+        }
+        
+        con.setReceiveStreamId(receiveId);
+        con.getPacketHandler().receivePacket(synPacket, con);
+        return con;
+    }
+    
+    /**
+     * Build a new connection to the given peer
+     */
+    public Connection connect(Destination peer, ConnectionOptions opts) {
+        Connection con = new Connection(_context, this, _schedulerChooser, _outboundQueue, _conPacketHandler, opts);
+        con.setRemotePeer(peer);
+        byte receiveId[] = new byte[4];
+        _context.random().nextBytes(receiveId);
+        synchronized (_connectionLock) {
+            ByteArray ba = new ByteArray(receiveId);
+            while (_connectionByInboundId.containsKey(ba)) {
+                _context.random().nextBytes(receiveId);
+            }
+            _connectionByInboundId.put(ba, con);
+        }
+        
+        con.setReceiveStreamId(receiveId);        
+        con.eventOccurred();
+        return con;
+    }
+
+    public MessageHandler getMessageHandler() { return _messageHandler; }
+    public PacketHandler getPacketHandler() { return _packetHandler; }
+    public ConnectionHandler getConnectionHandler() { return _connectionHandler; }
+    public I2PSession getSession() { return _session; }
+    public PacketQueue getPacketQueue() { return _outboundQueue; }
+    
+    /**
+     * Something b0rked hard, so kill all of our connections without mercy.
+     * Don't bother sending close packets.
+     *
+     */
+    public void disconnectAllHard() {
+        synchronized (_connectionLock) {
+            for (Iterator iter = _connectionByInboundId.values().iterator(); iter.hasNext(); ) {
+                Connection con = (Connection)iter.next();
+                con.disconnect(false);
+            }
+            _connectionByInboundId.clear();
+        }
+    }
+    
+    public void removeConnection(Connection con) {
+        synchronized (_connectionLock) {
+            _connectionByInboundId.remove(new ByteArray(con.getReceiveStreamId()));
+        }
+    }
+    
+    public Set listConnections() {
+        synchronized (_connectionLock) {
+            return new HashSet(_connectionByInboundId.values());
+        }
+    }
+    
+    public boolean ping(Destination peer, long timeoutMs) {
+        PingRequest req = new PingRequest();
+        byte id[] = new byte[4];
+        _context.random().nextBytes(id);
+        ByteArray ba = new ByteArray(id);
+        
+        synchronized (_pendingPings) {
+            _pendingPings.put(ba, req);
+        }
+        
+        PacketLocal packet = new PacketLocal(_context, peer);
+        packet.setSendStreamId(id);
+        packet.setFlag(Packet.FLAG_ECHO);
+        packet.setFlag(Packet.FLAG_SIGNATURE_INCLUDED);
+        packet.setOptionalFrom(_session.getMyDestination());
+        _outboundQueue.enqueue(packet);
+        
+        synchronized (req) {
+            if (!req.pongReceived())
+                try { req.wait(timeoutMs); } catch (InterruptedException ie) {}
+        }
+        
+        synchronized (_pendingPings) {
+            _pendingPings.remove(ba);
+        }
+        
+        boolean ok = req.pongReceived();
+        if (ok) {
+            _context.sessionKeyManager().tagsDelivered(peer.getPublicKey(), packet.getKeyUsed(), packet.getTagsSent());
+        }
+        return ok;
+    }
+    
+    private class PingRequest {
+        private boolean _ponged;
+        public PingRequest() { _ponged = false; }
+        public void pong() { 
+            synchronized (ConnectionManager.PingRequest.this) {
+                _ponged = true; 
+                ConnectionManager.PingRequest.this.notifyAll();
+            }
+        }
+        public boolean pongReceived() { return _ponged; }
+    }
+    
+    void receivePong(byte pingId[]) {
+        ByteArray ba = new ByteArray(pingId);
+        PingRequest req = null;
+        synchronized (_pendingPings) {
+            req = (PingRequest)_pendingPings.remove(ba);
+        }
+        if (req != null) 
+            req.pong();
+    }
+}

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

@@ -0,0 +1,145 @@
+package net.i2p.client.streaming;
+
+import java.util.Properties;
+
+/**
+ * Define the current options for the con (and allow custom tweaking midstream)
+ *
+ */
+public class ConnectionOptions extends I2PSocketOptions {
+    private int _connectDelay;
+    private boolean _fullySigned;
+    private int _windowSize;
+    private int _receiveWindow;
+    private int _profile;
+    private int _rtt;
+    private int _resendDelay;
+    private int _sendAckDelay;
+    private int _maxMessageSize;
+    private int _choke;
+    private int _maxResends;
+
+    public static final int PROFILE_BULK = 1;
+    public static final int PROFILE_INTERACTIVE = 2;
+    
+    public ConnectionOptions() {
+        super();
+        init(null);
+    }
+    
+    public ConnectionOptions(I2PSocketOptions opts) {
+        super(opts);
+        init(null);
+    }
+    
+    public ConnectionOptions(ConnectionOptions opts) {
+        super(opts);
+        init(opts);
+    }
+    
+    private void init(ConnectionOptions opts) {
+        if (opts != null) {
+            setConnectDelay(opts.getConnectDelay());
+            setProfile(opts.getProfile());
+            setRTT(opts.getRTT());
+            setRequireFullySigned(opts.getRequireFullySigned());
+            setWindowSize(opts.getWindowSize());
+            setResendDelay(opts.getResendDelay());
+            setMaxMessageSize(opts.getMaxMessageSize());
+            setChoke(opts.getChoke());
+            setMaxResends(opts.getMaxResends());
+        } else {
+            setConnectDelay(2*1000);
+            setProfile(PROFILE_BULK);
+            setMaxMessageSize(32*1024);
+            setRTT(5*1000);
+            setReceiveWindow(1);
+            setResendDelay(5*1000);
+            setSendAckDelay(1*1000);
+            setWindowSize(1);
+            setMaxResends(10);
+        }
+    }
+    
+    public ConnectionOptions(Properties opts) {
+        super(opts);
+        // load the options;
+    }
+    
+    /** 
+     * how long will we wait after instantiating a new con 
+     * before actually attempting to connect.  If this is
+     * set to 0, connect ASAP.  If it is greater than 0, wait
+     * until the output stream is flushed, the buffer fills, 
+     * or that many milliseconds pass.
+     *
+     */
+    public int getConnectDelay() { return _connectDelay; }
+    public void setConnectDelay(int delayMs) { _connectDelay = delayMs; }
+    
+    /**
+     * Do we want all packets in both directions to be signed,
+     * or can we deal with signatures on the SYN and FIN packets
+     * only?
+     *
+     */
+    public boolean getRequireFullySigned() { return _fullySigned; }
+    public void setRequireFullySigned(boolean sign) { _fullySigned = sign; }
+    
+    /** 
+     * How many messages will we send before waiting for an ACK?
+     *
+     */
+    public int getWindowSize() { return _windowSize; }
+    public void setWindowSize(int numMsgs) { _windowSize = numMsgs; }
+    
+    /** after how many consecutive messages should we ack? */
+    public int getReceiveWindow() { return _receiveWindow; } 
+    public void setReceiveWindow(int numMsgs) { _receiveWindow = numMsgs; }
+    
+    /**
+     * What to set the round trip time estimate to (in milliseconds)
+     */
+    public int getRTT() { return _rtt; }
+    public void setRTT(int ms) { _rtt = ms; }
+    
+    /** How long after sending a packet will we wait before resending? */
+    public int getResendDelay() { return _resendDelay; }
+    public void setResendDelay(int ms) { _resendDelay = ms; }
+    
+    /** 
+     * if there are packets we haven't ACKed yet and we don't 
+     * receive _receiveWindow messages before 
+     * (_lastSendTime+_sendAckDelay), send an ACK of what
+     * we have received so far.
+     *
+     */
+    public int getSendAckDelay() { return _sendAckDelay; }
+    public void setSendAckDelay(int delayMs) { _sendAckDelay = delayMs; }
+    
+    /** What is the largest message we want to send or receive? */
+    public int getMaxMessageSize() { return _maxMessageSize; }
+    public void setMaxMessageSize(int bytes) { _maxMessageSize = bytes; }
+    
+    /** 
+     * how long we want to wait before any data is transferred on the
+     * connection in either direction
+     *
+     */
+    public int getChoke() { return _choke; }
+    public void setChoke(int ms) { _choke = ms; }
+
+    /**
+     * What profile do we want to use for this connection?
+     *
+     */
+    public int getProfile() { return _profile; }
+    public void setProfile(int profile) { _profile = profile; }
+    
+    /**
+     * How many times will we try to send a message before giving up?
+     *
+     */
+    public int getMaxResends() { return _maxResends; }
+    public void setMaxResends(int numSends) { _maxResends = numSends; }
+}

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

@@ -0,0 +1,174 @@
+package net.i2p.client.streaming;
+
+import java.util.List;
+
+import net.i2p.I2PAppContext;
+import net.i2p.data.DataHelper;
+import net.i2p.data.Destination;
+import net.i2p.util.Log;
+
+/**
+ * Receive a packet for a particular connection - placing the data onto the
+ * queue, marking packets as acked, updating various fields, etc.
+ *
+ */
+public class ConnectionPacketHandler {
+    private I2PAppContext _context;
+    private Log _log;
+    
+    /** rtt = rtt*RTT_DAMPENING + (1-RTT_DAMPENING)*currentPacketRTT */
+    private static final double RTT_DAMPENING = 0.9;
+    
+    public ConnectionPacketHandler(I2PAppContext context) {
+        _context = context;
+        _log = context.logManager().getLog(ConnectionPacketHandler.class);
+    }
+    
+    /** distribute a packet to the connection specified */
+    void receivePacket(Packet packet, Connection con) {
+        boolean ok = verifyPacket(packet, con);
+        if (!ok) return;
+        boolean isNew = con.getInputStream().messageReceived(packet.getSequenceNum(), packet.getPayload());
+        if (isNew) {
+            con.incrementUnackedPacketsReceived();
+            long nextTime = con.getNextSendTime();
+            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);
+            } else {
+                if (_log.shouldLog(Log.DEBUG))
+                    _log.debug("Ack is already scheduled in " + nextTime + "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);
+            } else {
+                if (_log.shouldLog(Log.DEBUG))
+                    _log.debug("ACK only packet received: " + packet);
+            }
+        }
+        List acked = con.ackPackets(packet.getAckThrough(), packet.getNacks());
+        if ( (acked != null) && (acked.size() > 0) ) {
+            if (_log.shouldLog(Log.DEBUG))
+                _log.debug(acked.size() + " of our packets acked with " + packet);
+            // use the lowest RTT, since these would likely be bunched together,
+            // waiting for the most recent packet received before sending the ACK
+            int lowestRtt = -1;
+            for (int i = 0; i < acked.size(); i++) {
+                PacketLocal p = (PacketLocal)acked.get(i);
+                if ( (lowestRtt < 0) || (p.getAckTime() < lowestRtt) )
+                    lowestRtt = p.getAckTime();
+                
+                // ACK the tags we delivered so we can use them
+                if ( (p.getKeyUsed() != null) && (p.getTagsSent() != null) 
+                      && (p.getTagsSent().size() > 0) ) {
+                    _context.sessionKeyManager().tagsDelivered(p.getTo().getPublicKey(), 
+                                                               p.getKeyUsed(), 
+                                                               p.getTagsSent());
+                }
+                if (_log.shouldLog(Log.DEBUG))
+                    _log.debug("Packet acked: " + p);
+            }
+            int oldRTT = con.getOptions().getRTT();
+            int newRTT = (int)(RTT_DAMPENING*oldRTT + (1-RTT_DAMPENING)*lowestRtt);
+            con.getOptions().setRTT(newRTT);
+        }
+        
+        con.eventOccurred();
+    }
+    
+    /**
+     * Make sure this packet is ok and that we can continue processing its data.
+     * 
+     * @return true if the packet is ok for this connection, false if we shouldn't
+     *         continue processing.
+     */
+    private boolean verifyPacket(Packet packet, Connection con) {
+        if (packet.isFlagSet(Packet.FLAG_RESET)) {
+            verifyReset(packet, con);
+            return false;
+        } else {
+            boolean sigOk = verifySignature(packet, con);
+            
+            if (con.getSendStreamId() == null) {
+                if (packet.isFlagSet(Packet.FLAG_SYNCHRONIZE)) {
+                    con.setSendStreamId(packet.getReceiveStreamId());
+                    con.setRemotePeer(packet.getOptionalFrom());
+                    return true;
+                } else {
+                    // neither RST nor SYN and we dont have the stream id yet?  nuh uh
+                    if (_log.shouldLog(Log.WARN))
+                        _log.warn("Packet without RST or SYN where we dont know stream ID: " 
+                                  + packet);
+                    return false;
+                }
+            } else {
+                if (!DataHelper.eq(con.getSendStreamId(), packet.getReceiveStreamId())) {
+                    if (_log.shouldLog(Log.WARN))
+                        _log.warn("Packet received with the wrong reply stream id: " 
+                                  + con + " / " + packet);
+                    return false;
+                } else {
+                    return true;
+                }
+            }
+        }
+    }
+
+    /**
+     * Make sure this RST packet is valid, and if it is, act on it.
+     */
+    private void verifyReset(Packet packet, Connection con) {
+        if (DataHelper.eq(con.getReceiveStreamId(), packet.getSendStreamId())) {
+            boolean ok = packet.verifySignature(_context, packet.getOptionalFrom(), null);
+            if (!ok) {
+                if (_log.shouldLog(Log.WARN))
+                    _log.warn("Received unsigned / forged RST on " + con);
+                return;
+            } else {
+                if (_log.shouldLog(Log.DEBUG))
+                    _log.debug("Reset received");
+                // ok, valid RST
+                con.resetReceived();
+                con.eventOccurred();
+
+                // no further processing
+                return;
+            }
+        } else {
+            if (_log.shouldLog(Log.WARN))
+                _log.warn("Received a packet for the wrong connection?  wtf: " 
+                          + con + " / " + packet);
+            return;
+        }
+    }
+    
+    /**
+     * Verify the signature if necessary.  
+     *
+     * @return false only if the signature was required and it was invalid
+     */
+    private boolean verifySignature(Packet packet, Connection con) {
+        // verify the signature if necessary
+        if (con.getOptions().getRequireFullySigned() || 
+            packet.isFlagSet(Packet.FLAG_SYNCHRONIZE) ||
+            packet.isFlagSet(Packet.FLAG_CLOSE) ) {
+            // we need a valid signature
+            Destination from = con.getRemotePeer();
+            if (from == null)
+                from = packet.getOptionalFrom();
+            boolean sigOk = packet.verifySignature(_context, from, null);
+            if (!sigOk) {
+                if (_log.shouldLog(Log.WARN))
+                    _log.warn("Received unsigned / forged packet: " + packet);
+                return false;
+            }
+            if (packet.isFlagSet(Packet.FLAG_CLOSE))
+                con.closeReceived();
+        }
+        return true;
+    }    
+}

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

@@ -0,0 +1,24 @@
+package net.i2p.client.streaming;
+
+import java.net.ConnectException;
+import net.i2p.I2PException;
+
+/**
+ * Bridge to allow accepting new connections
+ *
+ */
+public class I2PServerSocketFull implements I2PServerSocket {
+    private I2PSocketManagerFull _socketManager;
+    
+    public I2PServerSocketFull(I2PSocketManagerFull mgr) {
+        _socketManager = mgr;
+    }
+    
+    public I2PSocket accept() throws I2PException {
+        return _socketManager.receiveSocket();
+    }
+    
+    public void close() { _socketManager.getConnectionManager().setAllowIncomingConnections(false); }
+    
+    public I2PSocketManager getManager() { return _socketManager; }
+}

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

@@ -0,0 +1,67 @@
+package net.i2p.client.streaming;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import net.i2p.data.Destination;
+
+/**
+ * Bridge between the full streaming lib and the I2PSocket API
+ *
+ */
+public class I2PSocketFull implements I2PSocket {
+    private Connection _connection;
+    private I2PSocket.SocketErrorListener _listener;
+    
+    public I2PSocketFull(Connection con) {
+        _connection = con;
+    }
+    
+    public void close() throws IOException {
+        if (_connection.getIsConnected()) {
+            _connection.disconnect(true);
+        } else {
+            throw new IOException("Not connected");
+        }
+    }
+    
+    public InputStream getInputStream() {
+        return _connection.getInputStream();
+    }
+    
+    public I2PSocketOptions getOptions() {
+        return _connection.getOptions();
+    }
+    
+    public OutputStream getOutputStream() throws IOException {
+        return _connection.getOutputStream();
+    }
+    
+    public Destination getPeerDestination() {
+        return _connection.getRemotePeer();
+    }
+    
+    public long getReadTimeout() {
+        return _connection.getOptions().getReadTimeout();
+    }
+    
+    public Destination getThisDestination() {
+        return _connection.getSession().getMyDestination();
+    }
+    
+    public void setOptions(I2PSocketOptions options) {
+        if (options instanceof ConnectionOptions)
+            _connection.setOptions((ConnectionOptions)options);
+        else
+            _connection.setOptions(new ConnectionOptions(options));
+    }
+    
+    public void setReadTimeout(long ms) {
+        _connection.getOptions().setReadTimeout(ms);
+    }
+    
+    public void setSocketErrorListener(I2PSocket.SocketErrorListener lsnr) {
+        _listener = lsnr;
+    }
+}

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

@@ -0,0 +1,175 @@
+package net.i2p.client.streaming;
+
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.io.UnsupportedEncodingException;
+import java.net.ConnectException;
+import java.net.NoRouteToHostException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Properties;
+import java.util.Set;
+
+import net.i2p.I2PAppContext;
+import net.i2p.I2PException;
+import net.i2p.client.I2PSession;
+import net.i2p.client.I2PSessionException;
+import net.i2p.client.I2PSessionListener;
+import net.i2p.data.Destination;
+import net.i2p.util.Log;
+
+
+/**
+ * Centralize the coordination and multiplexing of the local client's streaming.
+ * There should be one I2PSocketManager for each I2PSession, and if an application
+ * is sending and receiving data through the streaming library using an
+ * I2PSocketManager, it should not attempt to call I2PSession's setSessionListener
+ * or receive any messages with its .receiveMessage
+ *
+ */
+public class I2PSocketManagerFull implements I2PSocketManager {
+    private I2PAppContext _context;
+    private Log _log;
+    private I2PSession _session;
+    private I2PServerSocketFull _serverSocket;
+    private ConnectionOptions _defaultOptions;
+    private long _acceptTimeout;
+    private String _name;
+    private static int __managerId = 0;
+    private ConnectionManager _connectionManager;
+    
+    /**
+     * How long to wait for the client app to accept() before sending back CLOSE?
+     * This includes the time waiting in the queue.  Currently set to 5 seconds.
+     */
+    private static final long ACCEPT_TIMEOUT_DEFAULT = 5*1000;
+    
+    public I2PSocketManagerFull(I2PAppContext context, I2PSession session, Properties opts, String name) {
+        _context = context;
+        _session = session;
+        _log = _context.logManager().getLog(I2PSocketManagerFull.class);
+        _connectionManager = new ConnectionManager(_context, _session);
+        _name = name + " " + (++__managerId);
+        _acceptTimeout = ACCEPT_TIMEOUT_DEFAULT;
+        _defaultOptions = new ConnectionOptions(opts);
+        _serverSocket = new I2PServerSocketFull(this);
+    }
+
+    public I2PSession getSession() {
+        return _session;
+    }
+    
+    public ConnectionManager getConnectionManager() {
+        return _connectionManager;
+    }
+
+    public I2PSocket receiveSocket() throws I2PException {
+        if (_session.isClosed()) throw new I2PException("Session closed");
+        Connection con = _connectionManager.getConnectionHandler().accept(-1);
+        if (_log.shouldLog(Log.DEBUG))
+            _log.debug("receiveSocket() called: " + con);
+        if (con != null) {
+            I2PSocketFull sock = new I2PSocketFull(con);
+            con.setSocket(sock);
+            return sock;
+        } else { 
+            return null;
+        }
+    }
+    
+    /**
+     * Ping the specified peer, returning true if they replied to the ping within 
+     * the timeout specified, false otherwise.  This call blocks.
+     *
+     */
+    public boolean ping(Destination peer, long timeoutMs) {
+        return _connectionManager.ping(peer, timeoutMs);
+    }
+
+    /**
+     * How long should we wait for the client to .accept() a socket before
+     * sending back a NACK/Close?  
+     *
+     * @param ms milliseconds to wait, maximum
+     */
+    public void setAcceptTimeout(long ms) { _acceptTimeout = ms; }
+    public long getAcceptTimeout() { return _acceptTimeout; }
+
+    public void setDefaultOptions(I2PSocketOptions options) {
+        _defaultOptions = new ConnectionOptions(options);
+    }
+
+    public I2PSocketOptions getDefaultOptions() {
+        return _defaultOptions;
+    }
+
+    public I2PServerSocket getServerSocket() {
+        _connectionManager.setAllowIncomingConnections(true);
+        return _serverSocket;
+    }
+
+    /**
+     * Create a new connected socket (block until the socket is created)
+     *
+     * @param peer Destination to connect to
+     * @param options I2P socket options to be used for connecting
+     *
+     * @throws NoRouteToHostException if the peer is not found or not reachable
+     * @throws I2PException if there is some other I2P-related problem
+     */
+    public I2PSocket connect(Destination peer, I2PSocketOptions options) 
+                             throws I2PException, NoRouteToHostException {
+        if (_connectionManager.getSession().isClosed()) 
+            throw new I2PException("Session is closed");
+        Connection con = _connectionManager.connect(peer, new ConnectionOptions(options));
+        I2PSocketFull socket = new I2PSocketFull(con);
+        con.setSocket(socket);
+        if (con.getConnectionError() != null) { 
+            con.disconnect(false);
+            throw new NoRouteToHostException(con.getConnectionError());
+        }
+        return socket;
+    }
+
+    /**
+     * Create a new connected socket (block until the socket is created)
+     *
+     * @param peer Destination to connect to
+     *
+     * @throws NoRouteToHostException if the peer is not found or not reachable
+     * @throws I2PException if there is some other I2P-related problem
+     */
+    public I2PSocket connect(Destination peer) throws I2PException, NoRouteToHostException {
+        return connect(peer, _defaultOptions);
+    }
+
+    /**
+     * Destroy the socket manager, freeing all the associated resources.  This
+     * method will block untill all the managed sockets are closed.
+     *
+     */
+    public void destroySocketManager() {
+        _connectionManager.disconnectAllHard();
+        _connectionManager.setAllowIncomingConnections(false);
+        // should we destroy the _session too?
+    }
+
+    /**
+     * Retrieve a set of currently connected I2PSockets, either initiated locally or remotely.
+     *
+     */
+    public Set listSockets() {
+        Set connections = _connectionManager.listConnections();
+        Set rv = new HashSet(connections.size());
+        for (Iterator iter = connections.iterator(); iter.hasNext(); ) {
+            Connection con = (Connection)iter.next();
+            if (con.getSocket() != null)
+                rv.add(con.getSocket());
+        }
+        return rv;
+    }
+
+    public String getName() { return _name; }
+    public void setName(String name) { _name = name; }
+}

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

@@ -0,0 +1,78 @@
+package net.i2p.client.streaming;
+
+import net.i2p.I2PAppContext;
+import net.i2p.client.I2PSession;
+import net.i2p.client.I2PSessionListener;
+import net.i2p.client.I2PSessionException;
+import net.i2p.util.Log;
+
+/**
+ *
+ */
+public class MessageHandler implements I2PSessionListener {
+    private ConnectionManager _manager;
+    private I2PAppContext _context;
+    private Log _log;
+    
+    public MessageHandler(I2PAppContext ctx, ConnectionManager mgr) {
+        _manager = mgr;
+        _context = ctx;
+        _log = ctx.logManager().getLog(MessageHandler.class);
+    }
+        
+    /** Instruct the client that the given session has received a message with
+     * size # of bytes.
+     * @param session session to notify
+     * @param msgId message number available
+     * @param size size of the message
+     */
+    public void messageAvailable(I2PSession session, int msgId, long size) {
+        byte data[] = null;
+        try {
+            data = session.receiveMessage(msgId);
+        } catch (I2PSessionException ise) {
+            if (_log.shouldLog(Log.WARN))
+                _log.warn("Error receiving the message", ise);
+            return;
+        }
+        Packet packet = new Packet();
+        try {
+            packet.readPacket(data, 0, data.length);
+            _manager.getPacketHandler().receivePacket(packet);
+        } catch (IllegalArgumentException iae) {
+            if (_log.shouldLog(Log.WARN))
+                _log.warn("Received an invalid packet", iae);
+        }
+    }
+
+    /** Instruct the client that the session specified seems to be under attack
+     * and that the client may wish to move its destination to another router.
+     * @param session session to report abuse to
+     * @param severity how bad the abuse is
+     */
+    public void reportAbuse(I2PSession session, int severity) {
+        if (_log.shouldLog(Log.ERROR))
+            _log.error("Abuse reported with severity " + severity);
+        _manager.disconnectAllHard();
+    }
+
+    /**
+     * Notify the client that the session has been terminated
+     *
+     */
+    public void disconnected(I2PSession session) {
+        if (_log.shouldLog(Log.ERROR))
+            _log.error("I2PSession disconnected");
+        _manager.disconnectAllHard();
+    }
+
+    /**
+     * Notify the client that some error occurred
+     *
+     */
+    public void errorOccurred(I2PSession session, String message, Throwable error) {
+        if (_log.shouldLog(Log.ERROR))
+            _log.error("error occurred: " + message, error);
+        _manager.disconnectAllHard();
+    }
+}

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

@@ -11,7 +11,9 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 
+import net.i2p.I2PAppContext;
 import net.i2p.data.ByteArray;
+import net.i2p.util.Log;
 
 /**
  * Stream that can be given messages out of order 
@@ -19,6 +21,8 @@ import net.i2p.data.ByteArray;
  *
  */
 public class MessageInputStream extends InputStream {
+    private I2PAppContext _context;
+    private Log _log;
     /** 
      * List of ByteArray objects of data ready to be read,
      * with the first ByteArray at index 0, and the next
@@ -30,6 +34,8 @@ public class MessageInputStream extends InputStream {
     private int _readyDataBlockIndex;
     /** highest message ID used in the readyDataBlocks */
     private long _highestReadyBlockId;
+    /** highest overall message ID */
+    private long _highestBlockId;
     /** 
      * Message ID (Long) to ByteArray for blocks received
      * out of order when there are lower IDs not yet 
@@ -44,13 +50,17 @@ public class MessageInputStream extends InputStream {
     /** if we don't want any more data, ignore the data */
     private boolean _locallyClosed;
     private int _readTimeout;
+    private IOException _streamError;
     
     private Object _dataLock;
     
-    public MessageInputStream() {
+    public MessageInputStream(I2PAppContext ctx) {
+        _context = ctx;
+        _log = ctx.logManager().getLog(MessageInputStream.class);
         _readyDataBlocks = new ArrayList(4);
         _readyDataBlockIndex = 0;
         _highestReadyBlockId = -1;
+        _highestBlockId = -1;
         _readTimeout = -1;
         _notYetReadyBlocks = new HashMap(4);
         _dataLock = new Object();
@@ -65,6 +75,42 @@ public class MessageInputStream extends InputStream {
         }
     }
     
+    public long getHighestBlockId() { 
+        synchronized (_dataLock) {
+            return _highestBlockId;
+        }
+    }
+    
+    /**
+     * Retrieve the message IDs that are holes in our sequence - ones 
+     * past the highest ready ID and below the highest received message 
+     * ID.  This may return null if there are no such IDs.
+     *
+     */
+    public long[] getNacks() {
+        List ids = null;
+        synchronized (_dataLock) {
+            for (long i = _highestReadyBlockId + 1; i < _highestBlockId; i++) {
+                Long l = new Long(i);
+                if (_notYetReadyBlocks.containsKey(l)) {
+                    // ACK
+                } else {
+                    if (ids != null)
+                        ids = new ArrayList(4);
+                    ids.add(l);
+                }
+            }
+        }
+        if (ids != null) {
+            long rv[] = new long[ids.size()];
+            for (int i = 0; i < rv.length; i++)
+                rv[i] = ((Long)ids.get(i)).longValue();
+            return rv;
+        } else {
+            return null;
+        }
+    }
+    
     /**
      * Ascending list of block IDs greater than the highest
      * ready block ID, or null if there aren't any.
@@ -101,56 +147,110 @@ public class MessageInputStream extends InputStream {
     public int getReadTimeout() { return _readTimeout; }
     public void setReadTimeout(int timeout) { _readTimeout = timeout; }
     
+    public void closeReceived() {
+        synchronized (_dataLock) {
+            _closeReceived = true;
+        }
+    }
+    
     /**
      * A new message has arrived - toss it on the appropriate queue (moving 
-     * previously pending messages to the ready queue if it fills the gap, etc)
+     * previously pending messages to the ready queue if it fills the gap, etc).
      *
+     * @return true if this is a new packet, false if it is a dup
      */
-    public void messageReceived(long messageId, byte payload[]) {
+    public boolean messageReceived(long messageId, byte payload[]) {
         synchronized (_dataLock) {
-            if (messageId <= _highestReadyBlockId) return; // already received
+            if (_log.shouldLog(Log.DEBUG))
+                _log.debug("received " + messageId + " with " + payload.length);
+            if (messageId <= _highestReadyBlockId) {
+                if (_log.shouldLog(Log.DEBUG))
+                    _log.debug("ignoring dup message " + messageId);
+                return false; // already received
+            }
+            if (messageId > _highestBlockId)
+                _highestBlockId = messageId;
+            
             if (_highestReadyBlockId + 1 == messageId) {
-                if (!_locallyClosed)
+                if (!_locallyClosed && payload.length > 0) {
+                    if (_log.shouldLog(Log.DEBUG))
+                        _log.debug("accepting bytes as ready: " + payload.length);
                     _readyDataBlocks.add(new ByteArray(payload));
+                }
                 _highestReadyBlockId = messageId;
                 // now pull in any previously pending blocks
                 while (_notYetReadyBlocks.containsKey(new Long(_highestReadyBlockId + 1))) {
-                    _readyDataBlocks.add(_notYetReadyBlocks.get(new Long(_highestReadyBlockId + 1)));
+                    ByteArray ba = (ByteArray)_notYetReadyBlocks.get(new Long(_highestReadyBlockId + 1));
+                    if ( (ba != null) && (ba.getData() != null) && (ba.getData().length > 0) ) {
+                        _readyDataBlocks.add(ba);
+                    }
+                    
+                    if (_log.shouldLog(Log.DEBUG))
+                        _log.debug("making ready the block " + _highestReadyBlockId);
                     _highestReadyBlockId++;
                 }
                 _dataLock.notifyAll();
             } else {
+                if (_log.shouldLog(Log.DEBUG))
+                    _log.debug("message is out of order: " + messageId);
                 if (_locallyClosed) // dont need the payload, just the msgId in order
                     _notYetReadyBlocks.put(new Long(messageId), new ByteArray(null));
                 else
                     _notYetReadyBlocks.put(new Long(messageId), new ByteArray(payload));
             }
         }
+        return true;
     }
     
     public int read() throws IOException {
         if (_locallyClosed) throw new IOException("Already locally closed");
+        throwAnyError();
+        long expiration = -1;
+        if (_readTimeout > 0)
+            expiration = _readTimeout + System.currentTimeMillis();
         synchronized (_dataLock) {
-            if (_readyDataBlocks.size() <= 0) {
+            while (_readyDataBlocks.size() <= 0) {
+                if (_log.shouldLog(Log.DEBUG))
+                    _log.debug("read() with readyBlocks.size = " + _readyDataBlocks.size() + " on " + toString());
+                
                 if ( (_notYetReadyBlocks.size() <= 0) && (_closeReceived) ) {
+                    if (_log.shouldLog(Log.DEBUG))
+                        _log.debug("read() got EOF: " + toString());
                     return -1;
                 } else {
                     if (_readTimeout < 0) {
+                        if (_log.shouldLog(Log.DEBUG))
+                            _log.debug("read() with no timeout: " + toString());
                         try { _dataLock.wait(); } catch (InterruptedException ie) { }
+                        if (_log.shouldLog(Log.DEBUG))
+                            _log.debug("read() with no timeout complete: " + toString());
+                        throwAnyError();
                     } else if (_readTimeout > 0) {
+                        if (_log.shouldLog(Log.DEBUG))
+                            _log.debug("read() with timeout: " + _readTimeout + ": " + toString());
                         try { _dataLock.wait(_readTimeout); } catch (InterruptedException ie) { }
+                        if (_log.shouldLog(Log.DEBUG))
+                            _log.debug("read() with timeout complete: " + _readTimeout + ": " + toString());
+                        throwAnyError();
                     } else { // readTimeout == 0
                         // noop, don't block
+                        if (_log.shouldLog(Log.DEBUG))
+                            _log.debug("read() with nonblocking setup: " + toString());
                     }
                     if (_readyDataBlocks.size() <= 0) {
-                        throw new InterruptedIOException("Timeout reading");
+                        if ( (_readTimeout > 0) && (expiration > System.currentTimeMillis()) )
+                        throw new InterruptedIOException("Timeout reading (timeout=" + _readTimeout + ")");
                     }
                 }
             }
             
+            if (_log.shouldLog(Log.DEBUG))
+                _log.debug("read() readyBlocks = " + _readyDataBlocks.size() + ": " + toString());
+            
             // either was already ready, or we wait()ed and it arrived
             ByteArray cur = (ByteArray)_readyDataBlocks.get(0);
-            byte rv = cur.getData()[_readyDataBlockIndex++];
+            byte rv = cur.getData()[_readyDataBlockIndex];
+            _readyDataBlockIndex++;
             if (cur.getData().length <= _readyDataBlockIndex) {
                 _readyDataBlockIndex = 0;
                 _readyDataBlocks.remove(0);
@@ -161,6 +261,7 @@ public class MessageInputStream extends InputStream {
     
     public int available() throws IOException {
         if (_locallyClosed) throw new IOException("Already closed, you wanker");
+        throwAnyError();
         synchronized (_dataLock) {
             if (_readyDataBlocks.size() <= 0) 
                 return 0;
@@ -213,4 +314,23 @@ public class MessageInputStream extends InputStream {
             _locallyClosed = true;
         }
     }
+    
+    /**
+     * Stream b0rked, die with the given error
+     *
+     */
+    void streamErrorOccurred(IOException ioe) {
+        _streamError = ioe;
+        synchronized (_dataLock) {
+            _dataLock.notifyAll();
+        }
+    }
+    
+    private void throwAnyError() throws IOException {
+        if (_streamError != null) {
+            IOException ioe = _streamError;
+            _streamError = null;
+            throw ioe;
+        }
+    }
 }

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

@@ -3,24 +3,33 @@ package net.i2p.client.streaming;
 import java.io.IOException;
 import java.io.OutputStream;
 
+import net.i2p.I2PAppContext;
+import net.i2p.util.Log;
+
 /**
  *
  */
 public class MessageOutputStream extends OutputStream {
+    private I2PAppContext _context;
+    private Log _log;
     private byte _buf[];
     private int _valid;
     private Object _dataLock;
     private DataReceiver _dataReceiver;
     private IOException _streamError;
+    private boolean _closed;
     
-    public MessageOutputStream(DataReceiver receiver) {
+    public MessageOutputStream(I2PAppContext ctx, DataReceiver receiver) {
-        this(receiver, 64*1024);
+        this(ctx, receiver, 64*1024);
     }
-    public MessageOutputStream(DataReceiver receiver, int bufSize) {
+    public MessageOutputStream(I2PAppContext ctx, DataReceiver receiver, int bufSize) {
         super();
+        _context = ctx;
+        _log = ctx.logManager().getLog(MessageOutputStream.class);
         _buf = new byte[bufSize];
         _dataReceiver = receiver;
         _dataLock = new Object();
+        _closed = false;
     }
     
     public void write(byte b[]) throws IOException {
@@ -34,6 +43,7 @@ public class MessageOutputStream extends OutputStream {
                 if (_valid + remaining < _buf.length) {
                     // simply buffer the data, no flush
                     System.arraycopy(b, off, _buf, _valid, remaining);
+                    _valid += remaining;
                     remaining = 0;
                 } else {
                     // buffer whatever we can fit then flush,
@@ -43,7 +53,14 @@ public class MessageOutputStream extends OutputStream {
                     System.arraycopy(b, off, _buf, _valid, toWrite);
                     remaining -= toWrite;
                     _valid = _buf.length;
+                    if (_log.shouldLog(Log.DEBUG))
+                        _log.debug("write(b[], " + off + ", " + len + "): valid = " + _valid);
+                    // 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");
                     _valid = 0;
                     throwAnyError();
                 }
@@ -59,12 +76,26 @@ public class MessageOutputStream extends OutputStream {
     
     public void flush() throws IOException {
         synchronized (_dataLock) {
+            if (_log.shouldLog(Log.DEBUG))
+                _log.debug("flush(): valid = " + _valid);
+            // 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("flush(): valid = " + _valid + " complete");
             _valid = 0;
         }
         throwAnyError();
     }
     
+    public void close() throws IOException {
+        _closed = true;
+        flush();
+    }
+    
+    public boolean getClosed() { return _closed; }
+    
     private void throwAnyError() throws IOException {
         if (_streamError != null) {
             IOException ioe = _streamError;
@@ -82,7 +113,7 @@ public class MessageOutputStream extends OutputStream {
      * peer
      *
      */
-    void flushAvailable(DataReceiver target) {
+    void flushAvailable(DataReceiver target) throws IOException {
         synchronized (_dataLock) {
             target.writeData(_buf, 0, _valid);
             _valid = 0;
@@ -90,6 +121,6 @@ public class MessageOutputStream extends OutputStream {
     }
     
     public interface DataReceiver {
-        public void writeData(byte buf[], int off, int size);
+        public void writeData(byte buf[], int off, int size) throws IOException;
     }
 }

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

@@ -2,6 +2,7 @@ package net.i2p.client.streaming;
 
 import java.util.Arrays;
 import net.i2p.I2PAppContext;
+import net.i2p.data.Base64;
 import net.i2p.data.DataHelper;
 import net.i2p.data.Destination;
 import net.i2p.data.Signature;
@@ -42,6 +43,9 @@ import net.i2p.data.SigningPrivateKey;
  * to sign the entire header and payload with the space in the options 
  * for the signature being set to all zeroes.</p>
  *
+ * <p>If the sequenceNum is 0 and the SYN is not set, this is a plain ACK 
+ * packet that should not be ACKed</p>
+ *
  */
 public class Packet {
     private byte _sendStreamId[];
@@ -64,11 +68,11 @@ public class Packet {
      * synchronize packet)
      *
      */
-    public static final byte RECEIVE_STREAM_ID_UNKNOWN[] = new byte[] { 0x00, 0x00, 0x00, 0x00 };
+    public static final byte STREAM_ID_UNKNOWN[] = new byte[] { 0x00, 0x00, 0x00, 0x00 };
     
     /**
      * This packet is creating a new socket connection (if the receiveStreamId
-     * is RECEIVE_STREAM_ID_UNKNOWN) or it is acknowledging a request to 
+     * is STREAM_ID_UNKNOWN) or it is acknowledging a request to 
      * create a connection and in turn is accepting the socket.
      *
      */
@@ -122,19 +126,44 @@ public class Packet {
      *
      */
     public static final int FLAG_PROFILE_INTERACTIVE = (1 << 8);
+    /**
+     * If set, this packet is a ping (if sendStreamId is set) or a 
+     * 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? */
-    public byte[] getSendStreamId() { return _sendStreamId; }
+    public byte[] getSendStreamId() { 
-    public void setSendStreamId(byte[] id) { _sendStreamId = id; }
+        if ( (_sendStreamId == null) || (DataHelper.eq(_sendStreamId, STREAM_ID_UNKNOWN)) )
+            return null;
+        else
+            return _sendStreamId; 
+    }
+    public void setSendStreamId(byte[] id) { 
+        _sendStreamId = id; 
+        if ( (id != null) && (DataHelper.eq(id, STREAM_ID_UNKNOWN)) )
+            _sendStreamId = null;
+    }
     
     /** 
      * what is the stream replies should be sent on?  if the 
      * connection is still being built, this should be 
-     * {@see #RECEIVE_STREAM_ID_UNKNOWN}.
+     * null.
      *
      */
-    public byte[] getReceiveStreamId() { return _receiveStreamId; }
+    public byte[] getReceiveStreamId() { 
-    public void setReceiveStreamId(byte[] id) { _receiveStreamId = id; }
+        if ( (_receiveStreamId == null) || (DataHelper.eq(_receiveStreamId, STREAM_ID_UNKNOWN)) )
+            return null;
+        else
+            return _receiveStreamId; 
+    }
+    public void setReceiveStreamId(byte[] id) { 
+        _receiveStreamId = id; 
+        if ( (id != null) && (DataHelper.eq(id, STREAM_ID_UNKNOWN)) )
+            _receiveStreamId = null;
+    }
     
     /** 0-indexed sequence number for this Packet in the sendStream */
     public long getSequenceNum() { return _sequenceNum; }
@@ -173,14 +202,27 @@ public class Packet {
     /** is a particular flag set on this packet? */
     public boolean isFlagSet(int flag) { return 0 != (_flags & flag); }
     public void setFlag(int flag) { _flags |= flag; }
+    public void setFlag(int flag, boolean set) { 
+        if (set)
+            _flags |= flag; 
+        else
+            _flags &= ~flag;
+    }
 
     /** the signature on the packet (only included if the flag for it is set) */
     public Signature getOptionalSignature() { return _optionSignature; }
-    public void setOptionalSignature(Signature sig) { _optionSignature = sig; }
+    public void setOptionalSignature(Signature sig) { 
+        setFlag(FLAG_SIGNATURE_INCLUDED, sig != null);
+        _optionSignature = sig; 
+    }
 
     /** the sender of the packet (only included if the flag for it is set) */
     public Destination getOptionalFrom() { return _optionFrom; }
-    public void setOptionalFrom(Destination from) { _optionFrom = from; }
+    public void setOptionalFrom(Destination from) { 
+        setFlag(FLAG_FROM_INCLUDED, from != null);
+        if (from == null) throw new RuntimeException("from is null!?");
+        _optionFrom = from; 
+    }
     
     /** 
      * How many milliseconds the sender of this packet wants the recipient
@@ -188,14 +230,20 @@ public class Packet {
      * set) 
      */
     public int getOptionalDelay() { return _optionDelay; }
-    public void setOptionalDelay(int delayMs) { _optionDelay = delayMs; }
+    public void setOptionalDelay(int delayMs) { 
+        setFlag(FLAG_DELAY_REQUESTED, delayMs > 0);
+        _optionDelay = delayMs; 
+    }
     
     /**
      * What is the largest payload the sender of this packet wants to receive?
      *
      */
     public int getOptionalMaxSize() { return _optionMaxSize; }
-    public void setOptionalMaxSize(int numBytes) { _optionMaxSize = numBytes; }
+    public void setOptionalMaxSize(int numBytes) { 
+        setFlag(FLAG_MAX_PACKET_SIZE_INCLUDED, numBytes > 0);
+        _optionMaxSize = numBytes; 
+    }
     
     /**
      * Write the packet to the buffer (starting at the offset) and return
@@ -212,13 +260,19 @@ public class Packet {
      */
     private int writePacket(byte buffer[], int offset, boolean includeSig) throws IllegalStateException {
         int cur = offset;
-        System.arraycopy(_sendStreamId, 0, buffer, cur, _sendStreamId.length);
+        if (_sendStreamId != null)
-        cur += _sendStreamId.length;
+            System.arraycopy(_sendStreamId, 0, buffer, cur, _sendStreamId.length);
-        System.arraycopy(_receiveStreamId, 0, buffer, cur, _receiveStreamId.length);
+        else
-        cur += _receiveStreamId.length;
+            System.arraycopy(STREAM_ID_UNKNOWN, 0, buffer, cur, STREAM_ID_UNKNOWN.length);
-        DataHelper.toLong(buffer, cur, 4, _sequenceNum);
         cur += 4;
-        DataHelper.toLong(buffer, cur, 4, _ackThrough);
+        if (_receiveStreamId != null)
+            System.arraycopy(_receiveStreamId, 0, buffer, cur, _receiveStreamId.length);
+        else
+            System.arraycopy(STREAM_ID_UNKNOWN, 0, buffer, cur, STREAM_ID_UNKNOWN.length);
+        cur += 4;
+        DataHelper.toLong(buffer, cur, 4, _sequenceNum > 0 ? _sequenceNum : 0);
+        cur += 4;
+        DataHelper.toLong(buffer, cur, 4, _ackThrough > 0 ? _ackThrough : 0);
         cur += 4;
         if (_nacks != null) {
             DataHelper.toLong(buffer, cur, 1, _nacks.length);
@@ -231,7 +285,7 @@ public class Packet {
             DataHelper.toLong(buffer, cur, 1, 0);
             cur++;
         }
-        DataHelper.toLong(buffer, cur, 1, _resendDelay);
+        DataHelper.toLong(buffer, cur, 1, _resendDelay > 0 ? _resendDelay : 0);
         cur++;
         DataHelper.toLong(buffer, cur, 2, _flags);
         cur += 2;
@@ -250,21 +304,21 @@ public class Packet {
         cur += 2;
         
         if (isFlagSet(FLAG_DELAY_REQUESTED)) {
-            DataHelper.toLong(buffer, cur, 1, _optionDelay);
+            DataHelper.toLong(buffer, cur, 1, _optionDelay > 0 ? _optionDelay : 0);
             cur++;
         }
         if (isFlagSet(FLAG_FROM_INCLUDED)) {
             cur += _optionFrom.writeBytes(buffer, cur);
         }
         if (isFlagSet(FLAG_MAX_PACKET_SIZE_INCLUDED)) {
-            DataHelper.toLong(buffer, cur, 2, _optionMaxSize);
+            DataHelper.toLong(buffer, cur, 2, _optionMaxSize > 0 ? _optionMaxSize : DEFAULT_MAX_SIZE);
             cur += 2;
         }
         if (isFlagSet(FLAG_SIGNATURE_INCLUDED)) {
             if (includeSig)
                 System.arraycopy(_optionSignature.getData(), 0, buffer, cur, Signature.SIGNATURE_BYTES);
             else // we're signing (or validating)
-                Arrays.fill(buffer, cur, Signature.SIGNATURE_BYTES, (byte)0x0);
+                Arrays.fill(buffer, cur, cur+Signature.SIGNATURE_BYTES, (byte)0x0);
             cur += Signature.SIGNATURE_BYTES;
         }
         
@@ -276,6 +330,42 @@ public class Packet {
         return cur - offset;
     }
     
+    
+    /**
+     * how large would this packet be if we wrote it
+     */
+    public int writtenSize() throws IllegalStateException {
+        int size = 0;
+        size += _sendStreamId.length;
+        size += _receiveStreamId.length;
+        size += 4; // sequenceNum
+        size += 4; // ackThrough
+        if (_nacks != null) {
+			size++; // nacks length
+            size += 4 * _nacks.length;
+        } else {
+            size++; // nacks length
+        }
+        size++; // resendDelay
+        size += 2; // flags
+
+        if (isFlagSet(FLAG_DELAY_REQUESTED))
+            size += 1;
+        if (isFlagSet(FLAG_FROM_INCLUDED))
+            size += _optionFrom.size();
+        if (isFlagSet(FLAG_MAX_PACKET_SIZE_INCLUDED))
+            size += 2;
+        if (isFlagSet(FLAG_SIGNATURE_INCLUDED))
+            size += Signature.SIGNATURE_BYTES;
+        
+        size += 2; // option size
+        
+        if (_payload != null) {
+            size += _payload.length;
+        }
+        
+        return size;
+    }
     /**
      * Read the packet from the buffer (starting at the offset) and return
      * the number of bytes read.
@@ -337,10 +427,10 @@ public class Packet {
             cur += 2;
         }
         if (isFlagSet(FLAG_SIGNATURE_INCLUDED)) {
-            Signature sig = new Signature();
+            _optionSignature = new Signature();
             byte buf[] = new byte[Signature.SIGNATURE_BYTES];
             System.arraycopy(buffer, cur, buf, 0, Signature.SIGNATURE_BYTES);
-            sig.setData(buf);
+            _optionSignature.setData(buf);
             cur += Signature.SIGNATURE_BYTES;
         }
     }
@@ -355,6 +445,8 @@ public class Packet {
         if (!isFlagSet(FLAG_SIGNATURE_INCLUDED)) return false;
         if (_optionSignature == null) return false;
         
+        if (buffer == null)
+            buffer = new byte[writtenSize()];
         int size = writePacket(buffer, 0, false);
         return ctx.dsa().verifySignature(_optionSignature, buffer, 0, size, from.getSigningPublicKey());
     }
@@ -386,4 +478,33 @@ public class Packet {
         System.arraycopy(_optionSignature.getData(), 0, buffer, signatureOffset, Signature.SIGNATURE_BYTES);
         return size;
     }
+    
+    public String toString() {
+        return "Packet " + _sequenceNum + " on " + toId(_sendStreamId) 
+               + "<-->" + toId(_receiveStreamId) + ": " + toFlagString() 
+               + " ACK through " + _ackThrough 
+               + " size: " + (_payload != null ? _payload.length : 0);
+    }
+    
+    private static final String toId(byte id[]) {
+        if (id == null)
+            return Base64.encode(STREAM_ID_UNKNOWN);
+        else
+            return Base64.encode(id);
+    }
+    
+    private final String toFlagString() {
+        StringBuffer buf = new StringBuffer(32);
+        if (isFlagSet(FLAG_CLOSE)) buf.append(" CLOSE");
+        if (isFlagSet(FLAG_DELAY_REQUESTED)) buf.append(" DELAY");
+        if (isFlagSet(FLAG_ECHO)) buf.append(" ECHO");
+        if (isFlagSet(FLAG_FROM_INCLUDED)) buf.append(" FROM");
+        if (isFlagSet(FLAG_MAX_PACKET_SIZE_INCLUDED)) buf.append(" MAXSIZE");
+        if (isFlagSet(FLAG_PROFILE_INTERACTIVE)) buf.append(" INTERACTIVE");
+        if (isFlagSet(FLAG_RESET)) buf.append(" RESET");
+        if (isFlagSet(FLAG_SIGNATURE_INCLUDED)) buf.append(" SIG");
+        if (isFlagSet(FLAG_SIGNATURE_REQUESTED)) buf.append(" SIGREQ");
+        if (isFlagSet(FLAG_SYNCHRONIZE)) buf.append(" SYN");
+        return buf.toString();
+    }
 }

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

@@ -0,0 +1,155 @@
+package net.i2p.client.streaming;
+
+import java.util.Date;
+import java.util.Iterator;
+import java.util.Set;
+
+import net.i2p.I2PAppContext;
+import net.i2p.data.Base64;
+import net.i2p.data.DataHelper;
+import net.i2p.util.Log;
+
+/**
+ * receive a packet and dispatch it correctly to the connection specified,
+ * the server socket, or queue a reply RST packet.
+ *
+ */
+public class PacketHandler {
+    private ConnectionManager _manager;
+    private I2PAppContext _context;
+    private Log _log;
+    
+    public PacketHandler(I2PAppContext ctx, ConnectionManager mgr) {
+        _manager = mgr;
+        _context = ctx;
+        _log = ctx.logManager().getLog(PacketHandler.class);
+    }
+    
+    void receivePacket(Packet packet) {
+        if (_log.shouldLog(Log.DEBUG))
+            _log.debug("packet received: " + packet);
+        
+        byte sendId[] = packet.getSendStreamId();
+        if (!isNonZero(sendId))
+            sendId = null;
+        
+        Connection con = (sendId != null ? _manager.getConnectionByInboundId(sendId) : null); 
+        if (con != null) {
+            receiveKnownCon(con, packet);
+            System.out.println(new Date() + ": Receive packet " + packet + " on con " + con);
+        } else {
+            receiveUnknownCon(packet, sendId);
+            System.out.println(new Date() + ": Receive packet " + packet + " on an unknown con");
+        }
+    }
+    
+    private void receiveKnownCon(Connection con, Packet packet) {
+        // the packet is pointed at a stream ID we're receiving on
+        if (isValidMatch(con.getSendStreamId(), packet.getReceiveStreamId())) {
+            // the packet's receive stream ID also matches what we expect
+            if (_log.shouldLog(Log.DEBUG))
+                _log.debug("receive valid: " + packet);
+            con.getPacketHandler().receivePacket(packet, con);
+        } else {
+            if (packet.isFlagSet(Packet.FLAG_RESET)) {
+                // refused
+                if (_log.shouldLog(Log.DEBUG))
+                    _log.debug("receive reset: " + packet);
+                con.getPacketHandler().receivePacket(packet, con);
+            } else if (packet.isFlagSet(Packet.FLAG_SYNCHRONIZE)) {
+                if ( (con.getSendStreamId() == null) || 
+                     (DataHelper.eq(con.getSendStreamId(), packet.getReceiveStreamId())) ) {
+                    // con fully established, w00t
+                    con.setSendStreamId(packet.getReceiveStreamId());
+                    con.getPacketHandler().receivePacket(packet, con);
+                } else {
+                    if (_log.shouldLog(Log.WARN))
+                        _log.warn("Receive a syn packet with the wrong IDs: " + packet);
+                }
+            } else {
+                // someone is sending us a packet on the wrong stream 
+                if (_log.shouldLog(Log.WARN))
+                    _log.warn("Received a packet on the wrong stream: " + packet);
+            }
+        }
+    }
+    
+    private void receiveUnknownCon(Packet packet, byte sendId[]) {
+        if (packet.isFlagSet(Packet.FLAG_ECHO)) {
+            if (packet.getSendStreamId() != null) {
+                receivePing(packet);
+            } else if (packet.getReceiveStreamId() != null) {
+                receivePong(packet);
+            } else {
+                if (_log.shouldLog(Log.WARN))
+                    _log.warn("Echo packet received with no stream IDs: " + packet);
+            }
+        } else if (packet.isFlagSet(Packet.FLAG_SYNCHRONIZE)) {
+            if (sendId == null) {
+                // this is the initial SYN to establish a connection
+                _manager.getConnectionHandler().receiveNewSyn(packet);
+            } else {
+                if (_log.shouldLog(Log.WARN))
+                    _log.warn("Syn packet reply on a stream we don't know about: " + packet);
+            }
+        } else {
+            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());
+            }
+        }
+    }
+    
+    private void receivePing(Packet packet) {
+        boolean ok = packet.verifySignature(_context, packet.getOptionalFrom(), null);
+        if (!ok) {
+            if (_log.shouldLog(Log.WARN)) {
+                if (packet.getOptionalFrom() == null)
+                    _log.warn("Ping with no from (flagged? " + packet.isFlagSet(Packet.FLAG_FROM_INCLUDED) + ")");
+                else if (packet.getOptionalSignature() == null)
+                    _log.warn("Ping with no signature (flagged? " + packet.isFlagSet(Packet.FLAG_SIGNATURE_INCLUDED) + ")");
+                else
+                    _log.warn("Forged ping, discard (from=" + packet.getOptionalFrom().calculateHash().toBase64() 
+                              + " sig=" + packet.getOptionalSignature().toBase64() + ")");
+            }
+        } else {
+            PacketLocal pong = new PacketLocal(_context, packet.getOptionalFrom());
+            pong.setFlag(Packet.FLAG_ECHO, true);
+            pong.setFlag(Packet.FLAG_SIGNATURE_INCLUDED, false);
+            pong.setReceiveStreamId(packet.getSendStreamId());
+            _manager.getPacketQueue().enqueue(pong);
+        }
+    }
+    
+    private void receivePong(Packet packet) {
+        _manager.receivePong(packet.getReceiveStreamId());
+    }
+    
+    private static final boolean isValidMatch(byte conStreamId[], byte packetStreamId[]) {
+        if ( (conStreamId == null) || (packetStreamId == null) || 
+             (conStreamId.length != packetStreamId.length) ) 
+            return false;
+        
+        boolean nonZeroFound = false;
+        for (int i = 0; i < conStreamId.length; i++) {
+            if (conStreamId[i] != packetStreamId[i]) return false;
+            if (conStreamId[i] != 0x0) nonZeroFound = true;
+        }
+        return nonZeroFound;
+    }
+    
+    private static final boolean isNonZero(byte[] b) {
+        boolean nonZeroFound = false;
+        for (int i = 0; b != null && i < b.length; i++) {
+            if (b[i] != 0x0)
+                nonZeroFound = true;
+        }
+        return nonZeroFound;
+    }
+}

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

@@ -0,0 +1,63 @@
+package net.i2p.client.streaming;
+
+import java.util.Set;
+
+import net.i2p.I2PAppContext;
+import net.i2p.data.Destination;
+import net.i2p.data.SessionKey;
+
+/**
+ * coordinate local attributes about a packet - send time, ack time, number of
+ * retries, etc.
+ */
+public class PacketLocal extends Packet {
+    private I2PAppContext _context;
+    private Destination _to;
+    private SessionKey _keyUsed;
+    private Set _tagsSent;
+    private long _createdOn;
+    private int _numSends;
+    private long _lastSend;
+    private long _ackOn;
+    
+    public PacketLocal(I2PAppContext ctx, Destination to) {
+        _context = ctx;
+        _createdOn = ctx.clock().now();
+        _to = to;
+        _lastSend = -1;
+    }
+    
+    public Destination getTo() { return _to; }
+    public void setTo(Destination to) { _to = to; }
+    
+    public SessionKey getKeyUsed() { return _keyUsed; }
+    public void setKeyUsed(SessionKey key) { _keyUsed = key; }
+    
+    public Set getTagsSent() { return _tagsSent; }
+    public void setTagsSent(Set tags) { _tagsSent = tags; }
+    
+    public boolean shouldSign() { 
+        return isFlagSet(FLAG_SIGNATURE_INCLUDED) ||
+               isFlagSet(FLAG_SYNCHRONIZE) ||
+               isFlagSet(FLAG_CLOSE);
+    }
+    
+    public long getCreatedOn() { return _createdOn; }
+    public void incrementSends() { 
+        _numSends++;
+        _lastSend = _context.clock().now();
+    }
+    public void ackReceived() { 
+        if (_ackOn <= 0)
+            _ackOn = _context.clock().now(); 
+    }
+    /** how long after packet creation was it acked? */
+    public int getAckTime() {
+        if (_ackOn <= 0) 
+            return -1;
+        else
+            return (int)(_ackOn - _createdOn);
+    }
+    public int getNumSends() { return _numSends; }
+    public long getLastSend() { return _lastSend; }
+}

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

@@ -0,0 +1,56 @@
+package net.i2p.client.streaming;
+
+import java.util.Set;
+import java.util.HashSet;
+
+import net.i2p.I2PAppContext;
+import net.i2p.client.I2PSession;
+import net.i2p.client.I2PSessionException;
+import net.i2p.data.SessionKey;
+import net.i2p.util.Log;
+
+/**
+ *
+ */
+class PacketQueue {
+    private I2PAppContext _context;
+    private Log _log;
+    private I2PSession _session;
+    private byte _buf[];
+    
+    public PacketQueue(I2PAppContext context, I2PSession session) {
+        _context = context;
+        _session = session;
+        _buf = new byte[36*1024];
+        _log = context.logManager().getLog(PacketQueue.class);
+    }
+    
+    /**
+     * Add a new packet to be sent out ASAP
+     */
+    public void enqueue(PacketLocal packet) {
+        int size = 0;
+        if (packet.shouldSign())
+            size = packet.writeSignedPacket(_buf, 0, _context, _session.getPrivateKey());
+        else
+            size = packet.writePacket(_buf, 0);
+        
+        SessionKey keyUsed = new SessionKey();
+        Set tagsSent = new HashSet();
+        try {
+            // this should not block!
+            boolean sent = _session.sendMessage(packet.getTo(), _buf, 0, size, keyUsed, tagsSent);
+            if (!sent) {
+                if (_log.shouldLog(Log.WARN))
+                    _log.warn("Send failed for " + packet);
+            } else {
+                packet.setKeyUsed(keyUsed);
+                packet.setTagsSent(tagsSent);
+                packet.incrementSends();
+            }
+        } catch (I2PSessionException ise) {
+            if (_log.shouldLog(Log.WARN))
+                _log.warn("Unable to send the packet " + packet, ise);
+        }
+    }
+}

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

@@ -0,0 +1,62 @@
+package net.i2p.client.streaming;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import net.i2p.I2PAppContext;
+import net.i2p.util.Log;
+
+/**
+ * Examine a connection's state and pick the right scheduler for it.
+ *
+ */
+class SchedulerChooser {
+    private I2PAppContext _context;
+    private Log _log;
+    private TaskScheduler _nullScheduler;
+    /** list of TaskScheduler objects */
+    private List _schedulers;
+    
+    public SchedulerChooser(I2PAppContext context) {
+        _context = context;
+        _log = context.logManager().getLog(SchedulerChooser.class);
+        _schedulers = createSchedulers();
+        _nullScheduler = new NullScheduler();
+    }
+    
+    public TaskScheduler getScheduler(Connection con) {
+        for (int i = 0; i < _schedulers.size(); i++) {
+            TaskScheduler scheduler = (TaskScheduler)_schedulers.get(i);
+            if (scheduler.accept(con)) {
+                if (_log.shouldLog(Log.DEBUG))
+                    _log.debug("Scheduling for " + con + " with " + scheduler.getClass().getName());
+                return scheduler;
+            }
+        }
+        return _nullScheduler;
+    }
+    
+    private List createSchedulers() {
+        List rv = new ArrayList(8);
+        rv.add(new SchedulerPreconnect(_context));
+        rv.add(new SchedulerConnecting(_context));
+        rv.add(new SchedulerReceived(_context));
+        rv.add(new SchedulerConnectedBulk(_context));
+        rv.add(new SchedulerClosing(_context));
+        rv.add(new SchedulerClosed(_context));
+        rv.add(new SchedulerDead(_context));
+        return rv;
+    }
+    private class NullScheduler implements TaskScheduler {
+        private Log _log;
+        public NullScheduler() {
+            _log = _context.logManager().getLog(NullScheduler.class);
+        }
+        
+        public void eventOccurred(Connection con) {
+            if (_log.shouldLog(Log.DEBUG))
+                _log.debug("Event occurred on " + con, new Exception("source"));
+        }
+        public boolean accept(Connection con) { return true; }
+    };
+}

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

@@ -0,0 +1,51 @@
+package net.i2p.client.streaming;
+
+import net.i2p.I2PAppContext;
+import net.i2p.util.Log;
+
+/**
+ * <p>Scheduler used for after both sides have had their close packets
+ * ACKed, but the final timeout hasn't passed.</p>
+ *
+ * <h2>Entry conditions:</h2><ul>
+ * <li>Both sides have closed and ACKed.</li>
+ * <li>Less than the final timeout period has passed since the last ACK.</li>
+ * </ul>
+ *
+ * <h2>Events:</h2><ul>
+ * <li>Packets received</li>
+ * <li>RESET received</li>
+ * <li>Message sending fails (error talking to the session)</li>
+ * </ul>
+ *
+ * <h2>Next states:</h2>
+ * <li>{@link SchedulerDead dead} - after the final timeout passes</li>
+ * </ul>
+ *
+ *
+ */
+class SchedulerClosed extends SchedulerImpl {
+    private Log _log;
+    public SchedulerClosed(I2PAppContext ctx) {
+        super(ctx);
+        _log = ctx.logManager().getLog(SchedulerClosed.class);
+    }
+    
+    static final long CLOSE_TIMEOUT = 30*1000;
+    
+    public boolean accept(Connection con) {
+        boolean ok = (con != null) && 
+                     (con.getCloseSentOn() > 0) &&
+                     (con.getCloseReceivedOn() > 0) &&
+                     (con.getUnackedPacketsReceived() <= 0) &&
+                     (con.getUnackedPacketsSent() <= 0) &&
+                     (!con.getResetReceived()) &&
+                     (con.getCloseSentOn() + CLOSE_TIMEOUT > _context.clock().now());
+        return ok;
+    }
+    
+    public void eventOccurred(Connection con) {
+        long timeLeft = con.getCloseSentOn() + CLOSE_TIMEOUT - _context.clock().now();
+        reschedule(timeLeft, con);
+    }
+}

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

@@ -0,0 +1,53 @@
+package net.i2p.client.streaming;
+
+import net.i2p.I2PAppContext;
+import net.i2p.util.Log;
+
+/**
+ * <p>Scheduler used for after both SYNs have been ACKed and both sides 
+ * have closed the stream, but either we haven't ACKed their close or
+ * they haven't ACKed ours.</p>
+ *
+ * <h2>Entry conditions:</h2><ul>
+ * <li>Both sides have closed.</li>
+ * <li>At least one direction has not ACKed the close.</li>
+ * </ul>
+ *
+ * <h2>Events:</h2><ul>
+ * <li>Packets received (which may or may not ACK the ones sent)</li>
+ * <li>RESET received</li>
+ * <li>Message sending fails (error talking to the session)</li>
+ * <li>Message sending fails (too many resends)</li>
+ * </ul>
+ *
+ * <h2>Next states:</h2>
+ * <li>{@link SchedulerClosed closed} - after both sending and receiving ACKs on the CLOSE</li>
+ * <li>{@link SchedulerDead dead} - after sending or receiving a RESET</li>
+ * </ul>
+ *
+ */
+class SchedulerClosing extends SchedulerImpl {
+    private Log _log;
+    public SchedulerClosing(I2PAppContext ctx) {
+        super(ctx);
+        _log = ctx.logManager().getLog(SchedulerClosing.class);
+    }
+    
+    public boolean accept(Connection con) {
+        boolean ok = (con != null) && 
+                     (con.getCloseSentOn() > 0) &&
+                     (con.getCloseReceivedOn() > 0) &&
+                     ( (con.getUnackedPacketsReceived() > 0) || (con.getUnackedPacketsSent() > 0) );
+        return ok;
+    }
+    
+    public void eventOccurred(Connection con) {
+        if (con.getNextSendTime() <= 0)
+            con.setNextSendTime(_context.clock().now() + con.getOptions().getSendAckDelay());
+        long remaining = con.getNextSendTime() - _context.clock().now();
+        if (remaining <= 0)
+            con.sendAvailable();
+        else
+            reschedule(remaining, con);
+    }
+}

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

@@ -0,0 +1,66 @@
+package net.i2p.client.streaming;
+
+import net.i2p.I2PAppContext;
+import net.i2p.util.Log;
+
+/**
+ * <p>Scheduler used for after our SYN has been sent and ACKed but one
+ * (or more) sides haven't closed the stream yet.  In addition, the 
+ * stream must be using the BULK profile, rather than the INTERACTIVE
+ * profile.</p>
+ *
+ * <h2>Entry conditions:</h2><ul>
+ * <li>Packets sent and ACKs received.</li>
+ * <li>At least one direction is not closed</li>
+ * </ul>
+ *
+ * <h2>Events:</h2><ul>
+ * <li>Packets received (which may or may not ACK the ones sent)</li>
+ * <li>Message flush (explicitly, through a full buffer, or stream closure)</li>
+ * <li>RESET received</li>
+ * <li>Message sending fails (error talking to the session)</li>
+ * <li>Message sending fails (too many resends)</li>
+ * </ul>
+ *
+ * <h2>Next states:</h2>
+ * <li>{@link SchedulerClosing closing} - after both sending and receiving a CLOSE</li>
+ * <li>{@link SchedulerClosed closed} - after both sending and receiving ACKs on the CLOSE</li>
+ * <li>{@link SchedulerDead dead} - after sending or receiving a RESET</li>
+ * </ul>
+ *
+ */
+class SchedulerConnectedBulk extends SchedulerImpl {
+    private Log _log;
+    public SchedulerConnectedBulk(I2PAppContext ctx) {
+        super(ctx);
+        _log = ctx.logManager().getLog(SchedulerConnectedBulk.class);
+    }
+    
+    public boolean accept(Connection con) {
+        boolean ok = (con != null) && 
+                     (con.getAckedPackets() > 0) &&
+                     (con.getOptions().getProfile() == ConnectionOptions.PROFILE_BULK) &&
+                     (!con.getResetReceived()) &&
+                     ( (con.getCloseSentOn() <= 0) || (con.getCloseReceivedOn() <= 0) );
+        if (!ok) {
+            if (_log.shouldLog(Log.DEBUG))
+                _log.debug("con: " + con + " closeSentOn: " + con.getCloseSentOn() 
+                           + " closeReceivedOn: " + con.getCloseReceivedOn());
+        }
+        return ok;
+    }
+    
+    public void eventOccurred(Connection con) {
+        if (con.getNextSendTime() <= 0) 
+            return;
+        
+        long timeTillSend = con.getNextSendTime() - _context.clock().now();
+        
+        if (timeTillSend <= 0) {
+            con.setNextSendTime(-1);
+            con.sendAvailable();
+        } else {
+            reschedule(timeTillSend, con);
+        }
+    }
+}

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

@@ -0,0 +1,74 @@
+package net.i2p.client.streaming;
+
+import net.i2p.I2PAppContext;
+import net.i2p.util.Log;
+
+/**
+ * <p>Scheduler used once we've sent our SYN but it hasn't been ACKed yet.
+ * This connection may or may not be locally created.</p>
+ *
+ * <h2>Entry conditions:</h2><ul>
+ * <li>Packets sent but none ACKed</li>
+ * </ul>
+ *
+ * <h2>Events:</h2><ul>
+ * <li>Packets received (which may or may not ACK the ones sent)</li>
+ * <li>Message flush (explicitly, through a full buffer, or stream closure)</li>
+ * <li>Connection establishment timeout</li>
+ * <li>RESET received</li>
+ * </ul>
+ *
+ * <h2>Next states:</h2>
+ * <li>{@link SchedulerConnected connected} - after receiving an ACK</li>
+ * <li>{@link SchedulerClosing closing} - after both sending and receiving a CLOSE</li>
+ * <li>{@link SchedulerClosed closed} - after both sending and receiving ACKs on the CLOSE</li>
+ * <li>{@link SchedulerDead dead} - after sending or receiving a RESET</li>
+ * </ul>
+ *
+ */
+class SchedulerConnecting extends SchedulerImpl {
+    private Log _log;
+    
+    public SchedulerConnecting(I2PAppContext ctx) {
+        super(ctx);
+        _log = ctx.logManager().getLog(SchedulerConnecting.class);
+    }
+    
+    public boolean accept(Connection con) {
+        return (con != null) && 
+               (con.getLastSendId() >= 0) &&
+               (con.getAckedPackets() <= 0) && 
+               (!con.getResetReceived());
+    }
+    
+    public void eventOccurred(Connection con) {
+        long waited = _context.clock().now() - con.getCreatedOn();
+        if ( (con.getOptions().getConnectTimeout() > 0) && 
+             (con.getOptions().getConnectTimeout() <= waited) ) {
+            con.setConnectionError("Timeout waiting for ack (waited " + waited + "ms)");
+            con.disconnect(false);
+            reschedule(0, con);
+            if (_log.shouldLog(Log.DEBUG))
+                _log.debug("waited too long: " + waited);
+            return;
+        } else {
+            if (con.getOptions().getConnectTimeout() > 0)
+                reschedule(con.getOptions().getConnectTimeout(), con);
+        }
+        /*        
+        long timeTillSend = con.getNextSendTime() - _context.clock().now();
+        if ( (timeTillSend <= 0) && (con.getNextSendTime() > 0) ) {
+            if (_log.shouldLog(Log.DEBUG))
+                _log.debug("send next on " + con);
+            con.sendAvailable();
+            con.setNextSendTime(-1);
+        } else {
+            if (con.getNextSendTime() > 0) {
+                if (_log.shouldLog(Log.DEBUG))
+                    _log.debug("time till send: " + timeTillSend + " on " + con);
+                reschedule(timeTillSend, con);
+            }
+        }
+        */
+    }
+}

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

@@ -0,0 +1,47 @@
+package net.i2p.client.streaming;
+
+import net.i2p.I2PAppContext;
+import net.i2p.util.Log;
+
+/**
+ * <p>Scheduler used for after the final timeout has passed or the
+ * connection was reset.</p>
+ *
+ * <h2>Entry conditions:</h2><ul>
+ * <li>Both sides have closed and ACKed and the timeout has passed. <br />
+ *     <b>or</b></li>
+ * <li>A RESET was received</li>
+ * </ul>
+ *
+ * <h2>Events:</h2><ul>
+ * <li>None</li>
+ * </ul>
+ *
+ * <h2>Next states:</h2>
+ * <li>None</li>
+ * </ul>
+ *
+ *
+ */
+class SchedulerDead extends SchedulerImpl {
+    private Log _log;
+    public SchedulerDead(I2PAppContext ctx) {
+        super(ctx);
+        _log = ctx.logManager().getLog(SchedulerDead.class);
+    }
+    
+    public boolean accept(Connection con) {
+        boolean ok = (con != null) && 
+                     (con.getResetReceived()) ||
+                     ((con.getCloseSentOn() > 0) &&
+                      (con.getCloseReceivedOn() > 0) &&
+                      (con.getUnackedPacketsReceived() <= 0) &&
+                      (con.getUnackedPacketsSent() <= 0) &&
+                      (con.getCloseSentOn() + SchedulerClosed.CLOSE_TIMEOUT <= _context.clock().now()));
+        return ok;
+    }
+    
+    public void eventOccurred(Connection con) {
+        con.disconnectComplete();
+    }
+}

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

@@ -0,0 +1,36 @@
+package net.i2p.client.streaming;
+
+import net.i2p.I2PAppContext;
+import net.i2p.util.SimpleTimer;
+import net.i2p.util.Log;
+
+/**
+ * Base scheduler
+ */
+abstract class SchedulerImpl implements TaskScheduler {
+    protected I2PAppContext _context;
+    private Log _log;
+    
+    public SchedulerImpl(I2PAppContext ctx) {
+        _context = ctx;
+        _log = ctx.logManager().getLog(SchedulerImpl.class);
+    }
+    
+    protected void reschedule(long msToWait, Connection con) {
+        SimpleTimer.getInstance().addEvent(new ConEvent(con), msToWait);
+    }
+    
+    private class ConEvent implements SimpleTimer.TimedEvent {
+        private Connection _connection;
+        private Exception _addedBy;
+        public ConEvent(Connection con) { 
+            _connection = con; 
+            _addedBy = new Exception("added by");
+        }
+        public void timeReached() {
+            if (_log.shouldLog(Log.DEBUG))
+                _log.debug("firing event on " + _connection, _addedBy);
+            _connection.eventOccurred(); 
+        }
+    }
+}

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

@@ -0,0 +1,54 @@
+package net.i2p.client.streaming;
+
+import net.i2p.I2PAppContext;
+import net.i2p.util.Log;
+
+/**
+ * <p>Scheduler used for locally created connections where we have not yet
+ * sent the initial SYN packet.</p>
+ *
+ * <h2>Entry conditions:</h2><ul>
+ * <li>Locally created</li>
+ * <li>No packets sent or received</li>
+ * </ul>
+ *
+ * <h2>Events:</h2><ul>
+ * <li>Message flush (explicitly, through a full buffer, or stream closure)</li>
+ * <li>Initial delay timeout (causing implicit flush of any data available)</li>
+ * </ul>
+ *
+ * <h2>Next states:</h2>
+ * <li>{@link SchedulerConnecting connecting} - after sending a packet</li>
+ * </ul>
+ */
+class SchedulerPreconnect extends SchedulerImpl {
+    private Log _log;
+    
+    public SchedulerPreconnect(I2PAppContext ctx) {
+        super(ctx);
+        _log = ctx.logManager().getLog(SchedulerPreconnect.class);
+    }
+    
+    public boolean accept(Connection con) {
+        return (con != null) && 
+               (con.getSendStreamId() == null) &&
+               (con.getLastSendId() < 0);
+    }
+    
+    public void eventOccurred(Connection con) {
+        if (con.getNextSendTime() < 0)
+            con.setNextSendTime(_context.clock().now() + con.getOptions().getConnectDelay());
+        
+        long timeTillSend = con.getNextSendTime() - _context.clock().now();
+        if (timeTillSend <= 0) {
+            if (_log.shouldLog(Log.DEBUG))
+                _log.debug("Send available for the SYN on " + con);
+            con.sendAvailable();
+            con.setNextSendTime(-1);
+        } else {
+            if (_log.shouldLog(Log.DEBUG))
+                _log.debug("Wait " + timeTillSend + " before sending the SYN on " + con);
+            reschedule(timeTillSend, con);
+        }
+    }
+}

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

@@ -0,0 +1,44 @@
+package net.i2p.client.streaming;
+
+import net.i2p.I2PAppContext;
+import net.i2p.util.Log;
+
+/**
+ * Scheduler used after receiving an inbound connection but before
+ * we have sent our own SYN.
+ *
+ */
+class SchedulerReceived extends SchedulerImpl {
+    private Log _log;
+    
+    public SchedulerReceived(I2PAppContext ctx) {
+        super(ctx);
+        _log = ctx.logManager().getLog(SchedulerReceived.class);
+    }
+    
+    public boolean accept(Connection con) {
+        return (con != null) && 
+               (con.getLastSendId() < 0) &&
+               (con.getSendStreamId() != null);
+    }
+    
+    public void eventOccurred(Connection con) {
+        if (con.getUnackedPacketsReceived() <= 0) {
+            if (_log.shouldLog(Log.WARN))
+                _log.warn("hmm, state is received, but no unacked packets received?");
+            return;
+        }
+        
+        long timeTillSend = con.getNextSendTime() - _context.clock().now();
+        if (timeTillSend <= 0) {
+            if (_log.shouldLog(Log.DEBUG))
+                _log.debug("received con... send a packet");
+            con.sendAvailable();
+            con.setNextSendTime(-1);
+        } else {
+            if (_log.shouldLog(Log.DEBUG))
+                _log.debug("received con... time till next send: " + timeTillSend);
+            reschedule(timeTillSend, con);
+        }
+    }
+}

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

@@ -0,0 +1,22 @@
+package net.i2p.client.streaming;
+
+/**
+ * Coordinates what we do 'next'.  The scheduler used by a connection is
+ * selected based upon its current state.
+ *
+ */
+interface TaskScheduler {
+    /**
+     * An event has occurred (timeout, message sent, or message received),
+     * so schedule what to do next based on our current state.
+     *
+     */
+    public void eventOccurred(Connection con);
+    
+    /** 
+     * Determine whether this scheduler is fit to operate against the 
+     * given connection
+     *
+     */ 
+    public boolean accept(Connection con);
+}

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

@@ -0,0 +1,127 @@
+package net.i2p.client.streaming;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+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.Destination;
+import net.i2p.util.Log;
+
+/**
+ *
+ */
+public class ConnectTest {
+    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(30*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);
+                    try { Thread.sleep(10*1000); } catch (InterruptedException ie) {}
+                    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");
+                socket.close();
+                _log.debug("socket closed");
+            } 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[]) {
+        ConnectTest ct = new ConnectTest();
+        ct.test();
+    }
+}

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

@@ -0,0 +1,179 @@
+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.Destination;
+import net.i2p.util.Log;
+
+/**
+ *
+ */
+public class EchoTest {
+    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[5];
+                    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: " + new String(buf));
+                            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++) {
+                    out.write("blah!".getBytes());
+                    _log.debug("client wrote a line");
+                    out.flush();
+                    _log.debug("client flushed");
+                    byte buf[] = new byte[5];
+                    
+                    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));
+                            buf[j] = (byte)(c & 0xFF);
+                        }
+                    }
+                    if (buf != null) {
+                        _log.debug("* client read: " + new String(buf));
+                    }
+                }
+                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[]) {
+        EchoTest et = new EchoTest();
+        et.test();
+    }
+}

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

@@ -25,7 +25,7 @@ public class MessageInputStreamTest {
         byte orig[] = new byte[32*1024];
         _context.random().nextBytes(orig);
         
-        MessageInputStream in = new MessageInputStream();
+        MessageInputStream in = new MessageInputStream(_context);
         for (int i = 0; i < 32; i++) {
             byte msg[] = new byte[1024];
             System.arraycopy(orig, i*1024, msg, 0, 1024);
@@ -50,7 +50,7 @@ public class MessageInputStreamTest {
         byte orig[] = new byte[32*1024];
         _context.random().nextBytes(orig);
         
-        MessageInputStream in = new MessageInputStream();
+        MessageInputStream in = new MessageInputStream(_context);
         ArrayList order = new ArrayList(32);
         for (int i = 0; i < 32; i++)
             order.add(new Integer(i));

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

@@ -0,0 +1,56 @@
+package net.i2p.client.streaming;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+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.Destination;
+import net.i2p.util.Log;
+
+/**
+ *
+ */
+public class PingTest {
+    public void test() {
+        try {
+            I2PAppContext context = I2PAppContext.getGlobalContext();
+            I2PSession session = createSession();
+            ConnectionManager mgr = new ConnectionManager(context, session);
+            Log log = context.logManager().getLog(PingTest.class);
+            for (int i = 0; i < 10; i++) {
+                log.debug("ping " + i);
+                long before = context.clock().now();
+                boolean ponged = mgr.ping(session.getMyDestination(), 2*1000);
+                long after = context.clock().now();
+                log.debug("ponged? " + ponged + " after " + (after-before) + "ms");
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+        try { Thread.sleep(30*1000); } catch (Exception 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) {
+            e.printStackTrace();
+            throw new RuntimeException("b0rk b0rk b0rk");
+        }
+    }
+    
+    public static void main(String args[]) {
+        PingTest pt = new PingTest();
+        pt.test();
+    }
+}