Changeset d96ddd1

Timestamp:

Jun 3, 2015 11:42:54 AM (5 years ago)

Author:

zzz <zzz@…>

Branches:

master

Children:

f57e37d

Parents:

7b711eb (diff), b5455cee (diff)
Note: this is a merge changeset, the changes displayed below correspond to the merge itself.
Use the (diff) links above to see all the changes relative to each parent.

Message:

propagate from branch 'i2p.i2p.zzz.sam' (head 68de14d0053dea374413f9e0419b1c0f7e9ec3af)

to branch 'i2p.i2p' (head 54f5dd288f7c0c5a50f7f63f911aec4008be27e2)

Files:

• apps/sam/java/src/net/i2p/sam/Handler.java (added)
• apps/sam/java/src/net/i2p/sam/SAMBridge.java (modified) (20 diffs)
• apps/sam/java/src/net/i2p/sam/SAMHandler.java (modified) (9 diffs)
• apps/sam/java/src/net/i2p/sam/SAMHandlerFactory.java (modified) (3 diffs)
• apps/sam/java/src/net/i2p/sam/SAMMessageSession.java (modified) (2 diffs)
• apps/sam/java/src/net/i2p/sam/SAMUtils.java (modified) (2 diffs)
• apps/sam/java/src/net/i2p/sam/SAMv1Handler.java (modified) (16 diffs)
• apps/sam/java/src/net/i2p/sam/SAMv2Handler.java (modified) (2 diffs)
• apps/sam/java/src/net/i2p/sam/SAMv3Handler.java (modified) (13 diffs)
• apps/sam/java/src/net/i2p/sam/SAMv3StreamSession.java (modified) (10 diffs)
• core/java/src/net/i2p/client/I2PSessionImpl.java (modified) (2 diffs)
• core/java/src/net/i2p/client/I2PSimpleSession.java (modified) (1 diff)
• router/java/src/net/i2p/router/client/ClientListenerRunner.java (modified) (1 diff)

apps/sam/java/src/net/i2p/sam/SAMBridge.java

@@ -20 +20 @@
 import java.nio.ByteBuffer;
 import java.util.Arrays;
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Properties;
+import java.util.Set;
 
 import net.i2p.I2PAppContext;
@@ -31 +35 @@
 import net.i2p.util.I2PAppThread;
 import net.i2p.util.Log;
+import net.i2p.util.PortMapper;
 
 /**
@@ -56 +61 @@
      */
     private final Map<String,String> nameToPrivKeys;
+    private final Set<Handler> _handlers;
 
     private volatile boolean acceptConnections = true;
@@ -96 +102 @@
         persistFilename = options.keyFile;
         nameToPrivKeys = new HashMap<String,String>(8);
+        _handlers = new HashSet<Handler>(8);
         this.i2cpProps = options.opts;
         _state = INITIALIZED;
@@ -125 +132 @@
         persistFilename = persistFile;
         nameToPrivKeys = new HashMap<String,String>(8);
+        _handlers = new HashSet<Handler>(8);
         loadKeys();
         try {
@@ -210 +218 @@
     
     /**
-     * Load up the keys from the persistFilename
-     *
+     * Load up the keys from the persistFilename.
+     * TODO use DataHelper
+     * TODO store in config dir, not base dir
      */
     private void loadKeys() {
@@ -219 +228 @@
             try {
                 br = new BufferedReader(new InputStreamReader(
-                        new FileInputStream(persistFilename)));
+                        new FileInputStream(persistFilename), "UTF-8"));
                 String line = null;
                 while ( (line = br.readLine()) != null) {
@@ -227 +236 @@
                     nameToPrivKeys.put(name, privKeys);
                 }
+                if (_log.shouldInfo())
+                    _log.info("Loaded " + nameToPrivKeys.size() + " private keys from " + persistFilename);
             } catch (FileNotFoundException fnfe) {
                 _log.warn("Key file does not exist at " + persistFilename);
@@ -238 +249 @@
     
     /**
-     * Store the current keys to disk in the location specified on creation
-     *
+     * Store the current keys to disk in the location specified on creation.
+     * TODO use DataHelper
+     * TODO store in config dir, not base dir
      */
     private void storeKeys() {
@@ -249 +261 @@
                     String name = entry.getKey();
                     String privKeys = entry.getValue();
-                    out.write(name.getBytes());
+                    out.write(name.getBytes("UTF-8"));
                     out.write('=');
-                    out.write(privKeys.getBytes());
+                    out.write(privKeys.getBytes("UTF-8"));
                     out.write('\n');
                 }
+                if (_log.shouldInfo())
+                    _log.info("Saved " + nameToPrivKeys.size() + " private keys to " + persistFilename);
             } catch (IOException ioe) {
                 _log.error("Error writing out the SAM keys to " + persistFilename, ioe);
@@ -262 +276 @@
     }
     
+    /**
+     * Handlers must call on startup
+     * @since 0.9.20
+     */
+    public void register(Handler handler) {
+        if (_log.shouldInfo())
+            _log.info("Register " + handler);
+        synchronized (_handlers) {
+            _handlers.add(handler);
+        }
+    }
+    
+    /**
+     * Handlers must call on stop
+     * @since 0.9.20
+     */
+    public void unregister(Handler handler) {
+        if (_log.shouldInfo())
+            _log.info("Unregister " + handler);
+        synchronized (_handlers) {
+            _handlers.remove(handler);
+        }
+    }
+
+    /**
+     * Stop all the handlers.
+     * @since 0.9.20
+     */
+    private void stopHandlers() {
+        List<Handler> handlers = null;
+        synchronized (_handlers) {
+            if (!_handlers.isEmpty()) {
+                handlers = new ArrayList<Handler>(_handlers);
+                _handlers.clear();
+            }
+        }
+        if (handlers != null) {
+            for (Handler handler : handlers) {
+                if (_log.shouldInfo())
+                    _log.info("Stopping " + handler);
+                handler.stopHandling();
+            }
+        }
+    }
+
     ////// begin ClientApp interface, use only if using correct construtor
 
@@ -271 +330 @@
             return;
         changeState(STARTING);
+        synchronized (_handlers) {
+            _handlers.clear();
+        }
         loadKeys();
         try {
@@ -286 +348 @@
 
     /**
-     *  Does NOT stop existing sessions.
+     *  As of 0.9.20, stops running handlers and sessions.
+     *
      *  @since 0.9.6
      */
@@ -294 +357 @@
         changeState(STOPPING);
         acceptConnections = false;
+        stopHandlers();
         if (_runner != null)
             _runner.interrupt();
         else
             changeState(STOPPED);
-        // TODO does not stop active connections / sessions
     }
 
@@ -376 +439 @@
      */
     private void startThread() {
-        I2PAppThread t = new I2PAppThread(this, "SAMListener");
+        I2PAppThread t = new I2PAppThread(this, "SAMListener " + _listenPort);
         if (Boolean.parseBoolean(System.getProperty("sam.shutdownOnOOM"))) {
             t.addOOMEventThreadListener(new I2PAppThread.OOMEventListener() {
@@ -488 +551 @@
         if (_mgr != null)
             _mgr.register(this);
+        I2PAppContext.getGlobalContext().portMapper().register(PortMapper.SVC_SAM, _listenPort);
         try {
             while (acceptConnections) {
@@ -496 +560 @@
                                + s.socket().getPort());
 
-                class HelloHandler implements Runnable {
-                        private final SocketChannel s;
-                        private final SAMBridge parent;
-
-                        HelloHandler(SocketChannel s, SAMBridge parent) { 
+                class HelloHandler implements Runnable, Handler {
+                    private final SocketChannel s;
+                    private final SAMBridge parent;
+
+                    HelloHandler(SocketChannel s, SAMBridge parent) { 
                                 this.s = s ;
                                 this.parent = parent ;
-                        }
-
-                        public void run() {
+                    }
+
+                    public void run() {
+                        parent.register(this);
                         try {
-                            SAMHandler handler = SAMHandlerFactory.createSAMHandler(s, i2cpProps);
+                            SAMHandler handler = SAMHandlerFactory.createSAMHandler(s, i2cpProps, parent);
                             if (handler == null) {
                                 if (_log.shouldLog(Log.DEBUG))
@@ -516 +581 @@
                                 return;
                             }
-                            handler.setBridge(parent);
                             handler.startHandling();
                         } catch (SAMException e) {
@@ -527 +591 @@
                             try { s.close(); } catch (IOException ioe) {}
                             _log.log(Log.CRIT, "Unexpected error handling SAM connection", ee);
-                        }                               
-                        }
+                        } finally {
+                            parent.unregister(this);
+                        }
+                    }
+
+                    /** @since 0.9.20 */
+                    public void stopHandling() {
+                        try { s.close(); } catch (IOException ioe) {}
+                    }
                 }
-                // TODO: Handler threads are not saved or tracked and cannot be stopped
-                new I2PAppThread(new HelloHandler(s,this), "HelloHandler").start();
+                new I2PAppThread(new HelloHandler(s,this), "SAM HelloHandler").start();
             }
             changeState(STOPPING);
@@ -547 +617 @@
                     serverSocket.close();
             } catch (IOException e) {}
+            I2PAppContext.getGlobalContext().portMapper().unregister(PortMapper.SVC_SAM);
+            stopHandlers();
             changeState(STOPPED);
         }

apps/sam/java/src/net/i2p/sam/SAMHandler.java

@@ -26 +26 @@
  * @author human
  */
-abstract class SAMHandler implements Runnable {
+abstract class SAMHandler implements Runnable, Handler {
 
     protected final Log _log;
 
-    protected I2PAppThread thread = null;
-    protected SAMBridge bridge = null;
+    protected I2PAppThread thread;
+    protected final SAMBridge bridge;
 
     private final Object socketWLock = new Object(); // Guards writings on socket
@@ -42 +42 @@
     protected final Properties i2cpProps;
 
-    private final Object stopLock = new Object();
-    private volatile boolean stopHandler;
+    protected final Object stopLock = new Object();
+    protected boolean stopHandler;
 
     /**
@@ -54 +54 @@
      * @throws IOException 
      */
-    protected SAMHandler(SocketChannel s,
-                         int verMajor, int verMinor, Properties i2cpProps) throws IOException {
+    protected SAMHandler(SocketChannel s, int verMajor, int verMinor,
+                         Properties i2cpProps, SAMBridge parent) throws IOException {
         _log = I2PAppContext.getGlobalContext().logManager().getLog(getClass());
         socket = s;
@@ -62 +62 @@
         this.verMinor = verMinor;
         this.i2cpProps = i2cpProps;
+        bridge = parent;
     }
 
@@ -69 +70 @@
      */
     public final void startHandling() {
-        thread = new I2PAppThread(this, "SAMHandler");
+        thread = new I2PAppThread(this, getClass().getSimpleName());
         thread.start();
     }
-    
-    public void setBridge(SAMBridge bridge) { this.bridge = bridge; }
-    
+
     /**
      * Actually handle the SAM protocol.
@@ -82 +81 @@
 
     /**
-     * Get the input stream of the socket connected to the SAM client
-     *
-     * @return input stream
-     * @throws IOException 
+     * Get the channel of the socket connected to the SAM client
+     *
+     * @return channel
      */
     protected final SocketChannel getClientSocket() {
@@ -157 +155 @@
 
     /**
-     * Stop the SAM handler
-     *
-     */
-    public final void stopHandling() {
+     * Stop the SAM handler, close the client socket,
+     * unregister with the bridge.
+     */
+    public void stopHandling() {
         synchronized (stopLock) {
             stopHandler = true;
         }
+        try {
+            closeClientSocket();
+        } catch (IOException e) {}
+        bridge.unregister(this);
     }
 
@@ -184 +186 @@
     @Override
     public final String toString() {
-        return ("SAM handler (class: " + this.getClass().getName()
+        return (this.getClass().getSimpleName()
                 + "; SAM version: " + verMajor + "." + verMinor
                 + "; client: "
@@ -191 +193 @@
     }
 
+    /**
+     * Register with the bridge, call handle(),
+     * unregister with the bridge.
+     */
     public final void run() {
-        handle();
+        bridge.register(this);
+        try {
+            handle();
+        } finally {
+            bridge.unregister(this);
+        }
     }
 }

apps/sam/java/src/net/i2p/sam/SAMHandlerFactory.java

@@ -39 +39 @@
      * @return A SAM protocol handler, or null if the client closed before the handshake
      */
-    public static SAMHandler createSAMHandler(SocketChannel s, Properties i2cpProps) throws SAMException {
+    public static SAMHandler createSAMHandler(SocketChannel s, Properties i2cpProps,
+                                              SAMBridge parent) throws SAMException {
         StringTokenizer tok;
         Log log = I2PAppContext.getGlobalContext().logManager().getLog(SAMHandlerFactory.class);
@@ -46 +47 @@
             Socket sock = s.socket();
             sock.setSoTimeout(HELLO_TIMEOUT);
+            sock.setKeepAlive(true);
             String line = DataHelper.readLine(sock.getInputStream());
             sock.setSoTimeout(0);
@@ -104 +106 @@
             switch (verMajor) {
             case 1:
-                handler = new SAMv1Handler(s, verMajor, verMinor, i2cpProps);
+                handler = new SAMv1Handler(s, verMajor, verMinor, i2cpProps, parent);
                 break;
             case 2:
-                handler = new SAMv2Handler(s, verMajor, verMinor, i2cpProps);
+                handler = new SAMv2Handler(s, verMajor, verMinor, i2cpProps, parent);
                 break;
             case 3:
-                handler = new SAMv3Handler(s, verMajor, verMinor, i2cpProps);
+                handler = new SAMv3Handler(s, verMajor, verMinor, i2cpProps, parent);
                 break;
             default:

apps/sam/java/src/net/i2p/sam/SAMMessageSession.java

@@ -73 +73 @@
         handler = new SAMMessageSessionHandler(destStream, props);
 
+        // FIXME don't start threads in constructors
         Thread t = new I2PAppThread(handler, "SAMMessageSessionHandler");
         t.start();
@@ -126 +127 @@
     /**
      * Close a SAM message-based session.
-     *
      */
     public void close() {

apps/sam/java/src/net/i2p/sam/SAMUtils.java

@@ -213 +213 @@
 
     /* Dump a Properties object in an human-readable form */
+/****
     private static String dumpProperties(Properties props) {
         StringBuilder builder = new StringBuilder();
@@ -232 +233 @@
         return builder.toString();
     }
+****/
     
 /****

apps/sam/java/src/net/i2p/sam/SAMv1Handler.java

@@ -61 +61 @@
      * @throws IOException 
      */
-    public SAMv1Handler(SocketChannel s, int verMajor, int verMinor) throws SAMException, IOException {
-        this(s, verMajor, verMinor, new Properties());
+    public SAMv1Handler(SocketChannel s, int verMajor, int verMinor,
+                        SAMBridge parent) throws SAMException, IOException {
+        this(s, verMajor, verMinor, new Properties(), parent);
     }
     /**
@@ -76 +77 @@
      * @throws IOException 
      */
-    public SAMv1Handler(SocketChannel s, int verMajor, int verMinor, Properties i2cpProps) throws SAMException, IOException {
-        super(s, verMajor, verMinor, i2cpProps);
+    public SAMv1Handler(SocketChannel s, int verMajor, int verMinor,
+                        Properties i2cpProps, SAMBridge parent) throws SAMException, IOException {
+        super(s, verMajor, verMinor, i2cpProps, parent);
         _id = __id.incrementAndGet();
         if (_log.shouldLog(Log.DEBUG))
             _log.debug("SAM version 1 handler instantiated");
 
-    if ( ! verifVersion() ) {
+        if ( ! verifVersion() ) {
             throw new SAMException("BUG! Wrong protocol version!");
         }
@@ -184 +186 @@
         } catch (IOException e) {
             if (_log.shouldLog(Log.DEBUG))
-                _log.debug("Caught IOException ("
-                       + e.getMessage() + ") for message [" + msg + "]", e);
+                _log.debug("Caught IOException for message [" + msg + "]", e);
         } catch (Exception e) {
             _log.error("Unexpected exception for message [" + msg + "]", e);
@@ -194 +195 @@
                 closeClientSocket();
             } catch (IOException e) {
-                _log.error("Error closing socket: " + e.getMessage());
+                if (_log.shouldWarn())
+                    _log.warn("Error closing socket", e);
             }
             if (getRawSession() != null) {
@@ -798 +800 @@
         if (getRawSession() == null) {
             _log.error("BUG! Received raw bytes, but session is null!");
-            throw new NullPointerException("BUG! RAW session is null!");
+            return;
         }
 
@@ -819 +821 @@
         if (getRawSession() == null) {
             _log.error("BUG! Got raw receiving stop, but session is null!");
-            throw new NullPointerException("BUG! RAW session is null!");
+            return;
         }
 
@@ -834 +836 @@
         if (getDatagramSession() == null) {
             _log.error("BUG! Received datagram bytes, but session is null!");
-            throw new NullPointerException("BUG! DATAGRAM session is null!");
+            return;
         }
 
@@ -856 +858 @@
         if (getDatagramSession() == null) {
             _log.error("BUG! Got datagram receiving stop, but session is null!");
-            throw new NullPointerException("BUG! DATAGRAM session is null!");
+            return;
         }
 
@@ -874 +876 @@
         {
             _log.error ( "BUG! Want to answer to stream SEND, but session is null!" );
-            throw new NullPointerException ( "BUG! STREAM session is null!" );
+            return;
         }
     
@@ -892 +894 @@
         {
             _log.error ( "BUG! Stream outgoing buffer is free, but session is null!" );
-            throw new NullPointerException ( "BUG! STREAM session is null!" );
+            return;
         }
     
@@ -905 +907 @@
         if (getStreamSession() == null) {
             _log.error("BUG! Received stream connection, but session is null!");
-            throw new NullPointerException("BUG! STREAM session is null!");
+            return;
         }
 
@@ -915 +917 @@
     }
 
+    /** @param msg may be null */
     public void notifyStreamOutgoingConnection ( int id, String result, String msg ) throws IOException
     {
@@ -920 +923 @@
         {
             _log.error ( "BUG! Received stream connection, but session is null!" );
-            throw new NullPointerException ( "BUG! STREAM session is null!" );
-        }
-
-        String msgString = "" ;
-
-        if ( msg != null ) msgString = " MESSAGE=\"" + msg + "\"";
-
+            return;
+        }
+
+        String msgString = createMessageString(msg);
         if ( !writeString ( "STREAM STATUS RESULT="
                         + result
@@ -936 +936 @@
         }
     }
+
+    /**
+     *  Create a string to be appended to a status.
+     *
+     *  @param msg may be null
+     *  @return non-null, "" if msg is null, MESSAGE=msg or MESSAGE="msg a b c"
+     *           with leading space if msg is non-null
+     *  @since 0.9.20
+     */
+    protected static String createMessageString(String msg) {
+        String rv;
+        if ( msg != null ) {
+            msg = msg.replace("\n", " ");
+            msg = msg.replace("\r", " ");
+            if (!msg.startsWith("\"")) {
+                msg = msg.replace("\"", "");
+                if (msg.contains("\"") || msg.contains("\t"))
+                    msg = '"' + msg + '"';
+            }
+            rv = " MESSAGE=\"" + msg + "\"";
+        } else {
+            rv = "";
+        }
+        return rv;
+    }
   
     public void receiveStreamBytes(int id, ByteBuffer data) throws IOException {
         if (getStreamSession() == null) {
             _log.error("Received stream bytes, but session is null!");
-            throw new NullPointerException("BUG! STREAM session is null!");
+            return;
         }
 
@@ -957 +982 @@
     }
 
+    /** @param msg may be null */
     public void notifyStreamDisconnection(int id, String result, String msg) throws IOException {
         if (getStreamSession() == null) {
             _log.error("BUG! Received stream disconnection, but session is null!");
-            throw new NullPointerException("BUG! STREAM session is null!");
-        }
-
-        // FIXME: msg should be escaped!
-        if (!writeString("STREAM CLOSED ID=" + id + " RESULT=" + result
-                         + (msg == null ? "" : (" MESSAGE=" + msg))
-                         + "\n")) {
+            return;
+        }
+
+        String msgString = createMessageString(msg);
+        if (!writeString("STREAM CLOSED ID=" + id + " RESULT=" + result + msgString + '\n')) {
             throw new IOException("Error notifying disconnection to SAM client");
         }
@@ -977 +1001 @@
         if (getStreamSession() == null) {
             _log.error("BUG! Got stream receiving stop, but session is null!");
-            throw new NullPointerException("BUG! STREAM session is null!");
+            return;
         }
 

apps/sam/java/src/net/i2p/sam/SAMv2Handler.java

@@ -35 +35 @@
                  * @param verMinor SAM minor version to manage
                  */
-                public SAMv2Handler ( SocketChannel s, int verMajor, int verMinor ) throws SAMException, IOException
+                public SAMv2Handler(SocketChannel s, int verMajor, int verMinor,
+                                     SAMBridge parent) throws SAMException, IOException
                 {
-                        this ( s, verMajor, verMinor, new Properties() );
+                        this(s, verMajor, verMinor, new Properties(), parent);
                 }
 
@@ -51 +52 @@
                  */
 
-                public SAMv2Handler ( SocketChannel s, int verMajor, int verMinor, Properties i2cpProps ) throws SAMException, IOException
+                public SAMv2Handler(SocketChannel s, int verMajor, int verMinor,
+                                    Properties i2cpProps, SAMBridge parent) throws SAMException, IOException
                 {
-                        super ( s, verMajor, verMinor, i2cpProps );
+                        super(s, verMajor, verMinor, i2cpProps, parent);
                 }
                 

apps/sam/java/src/net/i2p/sam/SAMv3Handler.java

@@ -49 +49 @@
         private Session session;
         public static final SessionsDB sSessionsHash = new SessionsDB();
-        private boolean stolenSocket;
-        private boolean streamForwardingSocket;
+        private volatile boolean stolenSocket;
+        private volatile boolean streamForwardingSocket;
 
         
@@ -68 +68 @@
          * @param verMinor SAM minor version to manage
          */
-        public SAMv3Handler ( SocketChannel s, int verMajor, int verMinor ) throws SAMException, IOException
-        {
-                this ( s, verMajor, verMinor, new Properties() );
+        public SAMv3Handler(SocketChannel s, int verMajor, int verMinor,
+                            SAMBridge parent) throws SAMException, IOException
+        {
+                this(s, verMajor, verMinor, new Properties(), parent);
         }
 
@@ -84 +85 @@
          */
 
-        public SAMv3Handler ( SocketChannel s, int verMajor, int verMinor, Properties i2cpProps ) throws SAMException, IOException
-        {
-                super ( s, verMajor, verMinor, i2cpProps );
+        public SAMv3Handler(SocketChannel s, int verMajor, int verMinor,
+                            Properties i2cpProps, SAMBridge parent) throws SAMException, IOException
+        {
+                super(s, verMajor, verMinor, i2cpProps, parent);
                 if (_log.shouldLog(Log.DEBUG))
                         _log.debug("SAM version 3 handler instantiated");
@@ -215 +217 @@
         }
         
+        /**
+         *  The values in the SessionsDB
+         */
         public static class SessionRecord
         {
@@ -267 +272 @@
         }
 
+        /**
+         *  basically a HashMap from String to SessionRecord
+         */
         public static class SessionsDB
         {
                 private static final long serialVersionUID = 0x1;
 
-                static class ExistingIdException   extends Exception {
+                static class ExistingIdException extends Exception {
                         private static final long serialVersionUID = 0x1;
                 }
@@ -285 +293 @@
                 }
 
+                /** @return success */
                 synchronized public boolean put( String nick, SessionRecord session )
                         throws ExistingIdException, ExistingDestException
@@ -306 +315 @@
                 }
 
+                /** @return true if removed */
                 synchronized public boolean del( String nick )
                 {
-                        SessionRecord rec = map.get(nick);
-                        
-                        if ( rec!=null ) {
-                                map.remove(nick);
-                                return true ;
-                        }
-                        else
-                                return false ;
-                }
+                        return map.remove(nick) != null;
+                }
+
                 synchronized public SessionRecord get(String nick)
                 {
@@ -333 +337 @@
         }
         
+        /**
+         *  For SAMv3StreamSession connect and accept
+         */
         public void stealSocket()
         {
                 stolenSocket = true ;
                 this.stopHandling();
+        }
+        
+        /**
+         *  For SAMv3StreamSession
+         *  @since 0.9.20
+         */
+        SAMBridge getBridge() {
+                return bridge;
         }
         
@@ -349 +364 @@
                 this.thread.setName("SAMv3Handler " + _id);
                 if (_log.shouldLog(Log.DEBUG))
-                        _log.debug("SAM handling started");
+                        _log.debug("SAMv3 handling started");
 
                 try {
@@ -423 +438 @@
                 } catch (IOException e) {
                         if (_log.shouldLog(Log.DEBUG))
-                                _log.debug("Caught IOException ("
-                                        + e.getMessage() + ") for message [" + msg + "]", e);
+                                _log.debug("Caught IOException for message [" + msg + "]", e);
                 } catch (Exception e) {
                         _log.error("Unexpected exception for message [" + msg + "]", e);
@@ -436 +450 @@
                                         closeClientSocket();
                                 } catch (IOException e) {
-                                        _log.error("Error closing socket: " + e.getMessage());
+                                        if (_log.shouldWarn())
+                                                _log.warn("Error closing socket", e);
                                 }
                         }
@@ -445 +460 @@
                                                 ((SAMv3StreamSession)streamSession).stopForwardingIncoming();
                                         } catch (SAMException e) {
-                                                _log.error("Error while stopping forwarding connections: " + e.getMessage());
+                                                if (_log.shouldWarn())
+                                                        _log.warn("Error while stopping forwarding connections", e);
                                         } catch (InterruptedIOException e) {
-                                                _log.error("Interrupted while stopping forwarding connections: " + e.getMessage());
+                                                if (_log.shouldWarn())
+                                                        _log.warn("Interrupted while stopping forwarding connections", e);
                                         }
                                 }
                         }
-                
-
-
                         die();
                 }
         }
 
-        protected void die() {
+        /**
+         * Stop the SAM handler, close the socket,
+         * unregister with the bridge.
+         *
+         * Overridden to not close the client socket if stolen.
+         *
+         * @since 0.9.20
+         */
+        @Override
+        public void stopHandling() {
+            synchronized (stopLock) {
+                stopHandler = true;
+            }
+            if (!stolenSocket) {
+                try {
+                    closeClientSocket();
+                } catch (IOException e) {}
+            }
+            bridge.unregister(this);
+        }
+
+        private void die() {
                 SessionRecord rec = null ;
                 
@@ -814 +849 @@
         
 
-        public void notifyStreamResult(boolean verbose, String result, String message) throws IOException
-    {
+        public void notifyStreamResult(boolean verbose, String result, String message) throws IOException {
                 if (!verbose) return ;
-                
-                String out = "STREAM STATUS RESULT="+result;
-                if (message!=null)
-                        out = out + " MESSAGE=\"" + message + "\"";
-                out = out + '\n';
+                String msgString = createMessageString(message);
+                String out = "STREAM STATUS RESULT=" + result + msgString + '\n';
         
-        if ( !writeString ( out ) )
-        {
-            throw new IOException ( "Error notifying connection to SAM client" );
-        }
-    }
+                if (!writeString(out)) {
+                        throw new IOException ( "Error notifying connection to SAM client" );
+                }
+        }
 
         public void notifyStreamIncomingConnection(Destination d) throws IOException {

apps/sam/java/src/net/i2p/sam/SAMv3StreamSession.java

@@ -75 +75 @@
             /**
              * Connect the SAM STREAM session to the specified Destination
+             * for a single connection, using the socket stolen from the handler.
              *
              * @param handler The handler that communicates with the requesting client
@@ -88 +89 @@
              */
             public void connect ( SAMv3Handler handler, String dest, Properties props ) 
-            throws I2PException, ConnectException, NoRouteToHostException, 
+                throws I2PException, ConnectException, NoRouteToHostException, 
                         DataFormatException, InterruptedIOException, IOException {
 
@@ -118 +119 @@
                 WritableByteChannel toI2P      = Channels.newChannel(i2ps.getOutputStream());
                 
-                (new Thread(rec.getThreadGroup(), new I2PAppThread(new Pipe(fromClient,toI2P, "SAMPipeClientToI2P"), "SAMPipeClientToI2P"), "SAMPipeClientToI2P")).start();
-                (new Thread(rec.getThreadGroup(), new I2PAppThread(new Pipe(fromI2P,toClient, "SAMPipeI2PToClient"), "SAMPipeI2PToClient"), "SAMPipeI2PToClient")).start();
-                
+                SAMBridge bridge = handler.getBridge();
+                (new Thread(rec.getThreadGroup(),
+                            new Pipe(fromClient, toI2P, bridge),
+                            "ConnectV3 SAMPipeClientToI2P")).start();
+                (new Thread(rec.getThreadGroup(),
+                            new Pipe(fromI2P, toClient, bridge),
+                            "ConnectV3 SAMPipeI2PToClient")).start();
             }
 
             /**
-             * Accept an incoming STREAM
+             * Accept a single incoming STREAM on the socket stolen from the handler.
              *
              * @param handler The handler that communicates with the requesting client
@@ -151 +156 @@
                 }
                 
-                        I2PSocket i2ps;
-                        i2ps = this.socketServer.accept();
+                I2PSocket i2ps = this.socketServer.accept();
 
                 synchronized( this.socketServerLock )
@@ -160 +164 @@
                 
                 SAMv3Handler.SessionRecord rec = SAMv3Handler.sSessionsHash.get(nick);
-                
-                if ( rec==null || i2ps==null ) throw new InterruptedIOException() ;
-                
-                        if (verbose)
-                                handler.notifyStreamIncomingConnection(i2ps.getPeerDestination()) ;
+
+                if ( rec==null || i2ps==null ) throw new InterruptedIOException() ;
+
+                if (verbose)
+                        handler.notifyStreamIncomingConnection(i2ps.getPeerDestination()) ;
 
                 handler.stealSocket() ;
@@ -172 +176 @@
                 WritableByteChannel toI2P      = Channels.newChannel(i2ps.getOutputStream());
                 
-                (new Thread(rec.getThreadGroup(), new I2PAppThread(new Pipe(fromClient,toI2P, "SAMPipeClientToI2P"), "SAMPipeClientToI2P"), "SAMPipeClientToI2P")).start();
-                (new Thread(rec.getThreadGroup(), new I2PAppThread(new Pipe(fromI2P,toClient, "SAMPipeI2PToClient"), "SAMPipeI2PToClient"), "SAMPipeI2PToClient")).start();             
+                SAMBridge bridge = handler.getBridge();
+                (new Thread(rec.getThreadGroup(),
+                            new Pipe(fromClient, toI2P, bridge),
+                            "AcceptV3 SAMPipeClientToI2P")).start();
+                (new Thread(rec.getThreadGroup(),
+                            new Pipe(fromI2P, toClient, bridge),
+                            "AcceptV3 SAMPipeI2PToClient")).start();
             }
 
@@ -211 +220 @@
                 
                 SocketForwarder forwarder = new SocketForwarder(host, port, this, verbose);
-                (new Thread(rec.getThreadGroup(), new I2PAppThread(forwarder, "SAMStreamForwarder"), "SAMStreamForwarder")).start();
+                (new Thread(rec.getThreadGroup(), forwarder, "SAMV3StreamForwarder")).start();
             }
             
-            private static class SocketForwarder extends Thread
+            private static class SocketForwarder implements Runnable
             {
                 private final String host;
@@ -255 +264 @@
                                 // build pipes between both sockets
                                 try {
+                                        clientServerSock.socket().setKeepAlive(true);
                                         if (this.verbose)
                                                 SAMv3Handler.notifyStreamIncomingConnection(
@@ -262 +272 @@
                                         WritableByteChannel toClient   = clientServerSock ;
                                         WritableByteChannel toI2P      = Channels.newChannel(i2ps.getOutputStream());
-                                        (new I2PAppThread(new Pipe(fromClient,toI2P, "SAMPipeClientToI2P"), "SAMPipeClientToI2P")).start();
-                                        (new I2PAppThread(new Pipe(fromI2P,toClient, "SAMPipeI2PToClient"), "SAMPipeI2PToClient")).start();
+                                        (new I2PAppThread(new Pipe(fromClient, toI2P, null),
+                                                          "ForwardV3 SAMPipeClientToI2P")).start();
+                                        (new I2PAppThread(new Pipe(fromI2P,toClient, null),
+                                                          "ForwardV3 SAMPipeI2PToClient")).start();
 
                                 } catch (IOException e) {
@@ -278 +290 @@
             }
 
-            private static class Pipe extends Thread
+            private static class Pipe implements Runnable, Handler
             {
                 private final ReadableByteChannel in  ;
                 private final WritableByteChannel out ;
                 private final ByteBuffer buf ;
-                
-                public Pipe(ReadableByteChannel in, WritableByteChannel out, String name)
-                {
-                        super(name);
+                private final SAMBridge bridge;
+                
+                /**
+                 *  @param bridge may be null
+                 */
+                public Pipe(ReadableByteChannel in, WritableByteChannel out, SAMBridge bridge)
+                {
                         this.in  = in ;
                         this.out = out ;
                         this.buf = ByteBuffer.allocate(BUFFER_SIZE) ;
-                }
-                
-                public void run()
-                {
-                        try {
-                                while (!Thread.interrupted() && (in.read(buf)>=0 || buf.position() != 0)) {
-                                         buf.flip();
-                                         out.write(buf);
-                                         buf.compact();
-                                }
-                        }
-                                catch (IOException e)
-                                {
-                                        this.interrupt();
+                        this.bridge = bridge;
+                }
+                
+                public void run() {
+                        if (bridge != null)
+                                bridge.register(this);
+                        try {
+                                while (!Thread.interrupted() && (in.read(buf)>=0 || buf.position() != 0)) {
+                                         buf.flip();
+                                         out.write(buf);
+                                         buf.compact();
                                 }
-                        try {
-                                in.close();
-                        }
-                        catch (IOException e) {}
-                        try {
-                                buf.flip();
-                                while (buf.hasRemaining())
-                                        out.write(buf);
-                        }
-                        catch (IOException e) {}
-                        try {
-                                out.close();
-                        }
-                        catch (IOException e) {}
-                }
+                        } catch (IOException ioe) {
+                                // ignore
+                        } finally {
+                                try {
+                                        in.close();
+                                } catch (IOException e) {}
+                                try {
+                                        buf.flip();
+                                        while (buf.hasRemaining()) {
+                                                out.write(buf);
+                                        }
+                                } catch (IOException e) {}
+                                try {
+                                        out.close();
+                                } catch (IOException e) {}
+                                if (bridge != null)
+                                        bridge.unregister(this);
+                        }
+                }
+
+                /**
+                 *  Handler interface
+                 *  @since 0.9.20
+                 */
+                public void stopHandling() {
+                        try {
+                                in.close();
+                        } catch (IOException e) {}
+                }
             }
             

core/java/src/net/i2p/client/I2PSessionImpl.java

@@ -494 +494 @@
                             I2PSSLSocketFactory fact = new I2PSSLSocketFactory(_context, false, "certificates/i2cp");
                             _socket = fact.createSocket(_hostname, _portNum);
+                            _socket.setKeepAlive(true);
                         } catch (GeneralSecurityException gse) {
                             IOException ioe = new IOException("SSL Fail");
@@ -501 +502 @@
                     } else {
                         _socket = new Socket(_hostname, _portNum);
+                        _socket.setKeepAlive(true);
                     }
                     // _socket.setSoTimeout(1000000); // Uhmmm we could really-really use a real timeout, and handle it.

core/java/src/net/i2p/client/I2PSimpleSession.java

@@ -90 +90 @@
                         _socket = new Socket(_hostname, _portNum);
                     }
+                    _socket.setKeepAlive(true);
                     OutputStream out = _socket.getOutputStream();
                     out.write(I2PClient.PROTOCOL_BYTE);

router/java/src/net/i2p/router/client/ClientListenerRunner.java

@@ -94 +94 @@
                             if (_log.shouldLog(Log.DEBUG))
                                 _log.debug("Connection received");
+                            socket.setKeepAlive(true);
                             runConnection(socket);
                         } else {