2005-01-05 jrandom

* Handle unexpected network read errors more carefully (thanks parg!)
    * Added more methods to partially compare (DataHelper) and display
      arrays (Base64.encode).
    * Exposed the AES encryptBlock/decryptBlock on the context.aes()
    * Be more generous on the throttle when just starting up the router
    * Fix a missing scheduled event in the streaming lib (caused after reset)
    * Add a new DisconnectListener on the I2PSocketManager to allow
      notification of session destruction.
    * Make sure our own router identity is valid, and if it isn't, build a new
      one and restart the router.  Alternately, you can run the Router with
      the single command line argument "rebuild" and it will do the same.

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

@@ -33,6 +33,7 @@ public class Connection {
     private long _lastSendId;
     private boolean _resetReceived;
     private boolean _resetSent;
+    private long _resetSentOn;
     private boolean _connected;
     private boolean _hardDisconnected;
     private MessageInputStream _inputStream;
@@ -113,6 +114,7 @@ public class Connection {
         _ackSinceCongestion = true;
         _connectLock = new Object();
         _activeResends = 0;
+        _resetSentOn = -1;
         _context.statManager().createRateStat("stream.con.windowSizeAtCongestion", "How large was our send window when we send a dup?", "Stream", new long[] { 60*1000, 10*60*1000, 60*60*1000 });
         _context.statManager().createRateStat("stream.chokeSizeBegin", "How many messages were outstanding when we started to choke?", "Stream", new long[] { 60*1000, 10*60*1000, 60*60*1000 });
         _context.statManager().createRateStat("stream.chokeSizeEnd", "How many messages were outstanding when we stopped being choked?", "Stream", new long[] { 60*1000, 10*60*1000, 60*60*1000 });
@@ -185,6 +187,8 @@ public class Connection {
      */
     void sendReset() {
         _resetSent = true;
+        if (_resetSentOn <= 0)
+            _resetSentOn = _context.clock().now();
         if ( (_remotePeer == null) || (_sendStreamId == null) ) return;
         PacketLocal reply = new PacketLocal(_context, _remotePeer);
         reply.setFlag(Packet.FLAG_RESET);
@@ -388,6 +392,7 @@ public class Connection {
     public boolean getIsConnected() { return _connected; }
     public boolean getHardDisconnected() { return _hardDisconnected; }
     public boolean getResetSent() { return _resetSent; }
+    public long getResetSentOn() { return _resetSentOn; }
 
     void disconnect(boolean cleanDisconnect) {
         disconnect(cleanDisconnect, true);

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

@@ -5,9 +5,11 @@ import java.io.InterruptedIOException;
 import java.io.UnsupportedEncodingException;
 import java.net.ConnectException;
 import java.net.NoRouteToHostException;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Properties;
 import java.util.Set;
 
@@ -225,4 +227,12 @@ public class I2PSocketManagerFull implements I2PSocketManager {
 
     public String getName() { return _name; }
     public void setName(String name) { _name = name; }
+    
+    
+    public void addDisconnectListener(DisconnectListener lsnr) { 
+        _connectionManager.getMessageHandler().addDisconnectListener(lsnr);
+    }
+    public void removeDisconnectListener(DisconnectListener lsnr) {
+        _connectionManager.getMessageHandler().removeDisconnectListener(lsnr);
+    }
 }

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

@@ -1,5 +1,8 @@
 package net.i2p.client.streaming;
 
+import java.util.ArrayList;
+import java.util.List;
+
 import net.i2p.I2PAppContext;
 import net.i2p.client.I2PSession;
 import net.i2p.client.I2PSessionListener;
@@ -15,10 +18,12 @@ public class MessageHandler implements I2PSessionListener {
     private ConnectionManager _manager;
     private I2PAppContext _context;
     private Log _log;
+    private List _listeners;
     
     public MessageHandler(I2PAppContext ctx, ConnectionManager mgr) {
         _manager = mgr;
         _context = ctx;
+        _listeners = new ArrayList(1);
         _log = ctx.logManager().getLog(MessageHandler.class);
         _context.statManager().createRateStat("stream.packetReceiveFailure", "When do we fail to decrypt or otherwise receive a packet sent to us?", "Stream", new long[] { 60*60*1000, 24*60*60*1000 });
     }
@@ -69,6 +74,16 @@ public class MessageHandler implements I2PSessionListener {
         if (_log.shouldLog(Log.ERROR))
             _log.error("I2PSession disconnected");
         _manager.disconnectAllHard();
+        
+        List listeners = null;
+        synchronized (_listeners) {
+            listeners = new ArrayList(_listeners);
+            _listeners.clear();
+        }
+        for (int i = 0; i < listeners.size(); i++) {
+            I2PSocketManager.DisconnectListener lsnr = (I2PSocketManager.DisconnectListener)listeners.get(i);
+            lsnr.sessionDisconnected();
+        }
     }
 
     /**
@@ -80,4 +95,15 @@ public class MessageHandler implements I2PSessionListener {
             _log.error("error occurred: " + message, error);
         //_manager.disconnectAllHard();
     }
+    
+    public void addDisconnectListener(I2PSocketManager.DisconnectListener lsnr) { 
+        synchronized (_listeners) {
+            _listeners.add(lsnr);
+        }
+    }
+    public void removeDisconnectListener(I2PSocketManager.DisconnectListener lsnr) {
+        synchronized (_listeners) {
+            _listeners.remove(lsnr);
+        }
+    }
 }

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

@@ -33,10 +33,13 @@ class SchedulerDead extends SchedulerImpl {
     public boolean accept(Connection con) {
         if (con == null) return false;
         long timeSinceClose = _context.clock().now() - con.getCloseSentOn();
+        if (con.getResetSent())
+            timeSinceClose = _context.clock().now() - con.getResetSentOn();
         boolean nothingLeftToDo = (con.getCloseSentOn() > 0) &&
                                   (con.getCloseReceivedOn() > 0) &&
                                   (con.getUnackedPacketsReceived() <= 0) &&
                                   (con.getUnackedPacketsSent() <= 0) &&
+                                  (con.getResetSent()) &&
                                   (timeSinceClose >= Connection.DISCONNECT_TIMEOUT);
         boolean timedOut = (con.getOptions().getConnectTimeout() < con.getLifetime()) && 
                            con.getSendStreamId() == null &&

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

@@ -34,6 +34,8 @@ class SchedulerHardDisconnected extends SchedulerImpl {
     public boolean accept(Connection con) {
         if (con == null) return false;
         long timeSinceClose = _context.clock().now() - con.getCloseSentOn();
+        if (con.getResetSent())
+            timeSinceClose = _context.clock().now() - con.getResetSentOn();
         boolean ok = (con.getHardDisconnected() || con.getResetSent()) && 
                      (timeSinceClose < Connection.DISCONNECT_TIMEOUT);
         return ok;