SAM v3.3 master sessions.

Compiles only. Untested, not regression tested, not complete.

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

@@ -0,0 +1,358 @@
+package net.i2p.sam;
+
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.net.ConnectException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.net.SocketTimeoutException;
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+
+import net.i2p.I2PException;
+import net.i2p.client.I2PSession;
+import net.i2p.client.I2PSessionException;
+import net.i2p.client.I2PSessionMuxedListener;
+import net.i2p.client.streaming.I2PSocket;
+import net.i2p.data.DataFormatException;
+import net.i2p.data.DataHelper;
+import net.i2p.data.Destination;
+import net.i2p.util.Log;
+
+/**
+ * A session that does nothing, but implements interfaces for raw, datagram, and streaming
+ * for convenience.
+ *
+ * We extend SAMv3StreamSession as we must have it set up the I2PSession, in case
+ * user adds a STREAM session (and he probably will).
+ * This session receives all data from I2P, but you can't send any data on it.
+ *
+ * @since 0.9.25
+ */
+class MasterSession extends SAMv3StreamSession implements SAMDatagramReceiver, SAMRawReceiver,
+                                                          SAMMessageSess, I2PSessionMuxedListener {
+	private final SAMv3Handler handler;
+	private final SAMv3DatagramServer dgs;
+	private final Map<String, SAMMessageSess> sessions;
+
+	/**
+	 * Build a Session according to information
+	 * registered with the given nickname
+	 *   
+	 * @param nick nickname of the session
+	 * @throws IOException
+	 * @throws DataFormatException
+	 * @throws I2PSessionException
+	 */
+	public MasterSession(String nick, SAMv3DatagramServer dgServer, SAMv3Handler handler, Properties props) 
+			throws IOException, DataFormatException, SAMException {
+		super(nick);
+		props.setProperty("net.i2p.streaming.enforceProtocol", "true");
+                props.setProperty("i2cp.dontPublishLeaseSet", "false");
+		props.setProperty("FROM_PORT", Integer.toString(I2PSession.PORT_UNSPECIFIED));
+		props.setProperty("TO_PORT", Integer.toString(I2PSession.PORT_UNSPECIFIED));
+		dgs = dgServer;
+		sessions = new ConcurrentHashMap<String, SAMMessageSess>(4);
+		this.handler = handler;
+		I2PSession isess = socketMgr.getSession();
+		// if we get a RAW session added with 0/0, it will replace this,
+		// and we won't add this back if removed.
+		isess.addMuxedSessionListener(this, I2PSession.PROTO_ANY, I2PSession.PORT_ANY);
+	}
+
+	/**
+	 *  Add a session
+	 *  @return null for success, or error message
+	 */
+	public synchronized String add(String nick, String style, Properties props) {
+		SessionRecord rec = SAMv3Handler.sSessionsHash.get(nick);
+		if (rec != null || sessions.containsKey(nick))
+			return "Duplicate ID " + nick;
+		int listenPort = I2PSession.PORT_ANY;
+		String slp = (String) props.remove("LISTEN_PORT");
+                if (slp == null)
+                    slp = props.getProperty("FROM_PORT");
+		if (slp != null) {
+			try {
+				listenPort = Integer.parseInt(slp);
+				if (listenPort < 0 || listenPort > 65535)
+					return "Bad LISTEN_PORT " + slp;
+				// TODO enforce streaming listen port must be 0 or from port
+			} catch (NumberFormatException nfe) {
+				return "Bad LISTEN_PORT " + slp;
+			}
+		}
+		int listenProtocol;
+		SAMMessageSess sess;
+		// temp
+		try {
+			I2PSession isess = socketMgr.getSession();
+			if (style.equals("RAW")) {
+				if (!props.containsKey("PORT"))
+					return "RAW subsession must specify PORT";
+				listenProtocol = I2PSession.PROTO_DATAGRAM_RAW;
+				String spr = (String) props.remove("LISTEN_PROTOCOL");
+	                        if (spr == null)
+	                            spr = props.getProperty("PROTOCOL");
+				if (spr != null) {
+					try {
+						listenProtocol = Integer.parseInt(spr);
+						// RAW can't listen on streaming protocol
+						if (listenProtocol < 0 || listenProtocol > 255 ||
+						    listenProtocol == I2PSession.PROTO_STREAMING)
+							return "Bad RAW LISTEN_PPROTOCOL " + spr;
+					} catch (NumberFormatException nfe) {
+						return "Bad LISTEN_PROTOCOL " + spr;
+					}
+				}
+				sess = new SAMv3RawSession(nick, props, handler, isess, listenProtocol, listenPort, dgs);
+			} else if (style.equals("DATAGRAM")) {
+				if (!props.containsKey("PORT"))
+					return "DATAGRAM subsession must specify PORT";
+				listenProtocol = I2PSession.PROTO_DATAGRAM;
+				sess = new SAMv3DatagramSession(nick, props, handler, isess, listenPort, dgs);
+			} else if (style.equals("STREAM")) {
+				listenProtocol = I2PSession.PROTO_STREAMING;
+				// FIXME need something that hangs off an existing dest
+				sess = new SAMv3StreamSession(nick, props, handler, socketMgr, listenPort);
+			} else {
+				return "Unrecognized SESSION STYLE " + style;
+			}
+		} catch (Exception e) {
+			// temp
+			return e.toString();
+		}
+
+		for (SAMMessageSess s : sessions.values()) {
+			if (listenProtocol == s.getListenProtocol() &&
+			    listenPort == s.getListenPort())
+				return "Duplicate protocol " + listenProtocol + " and port " + listenPort;
+		}
+
+		// add to session db and our map
+		rec = new SessionRecord(getDestination().toBase64(), props, handler);
+		try {
+			if (!SAMv3Handler.sSessionsHash.put(nick, rec))
+				return "Duplicate ID " + nick;
+			sessions.put(nick, sess);
+		} catch (SessionsDB.ExistingIdException e) {
+			return e.toString();
+		} catch (SessionsDB.ExistingDestException e) {
+			// fixme need new db method for dup dests
+		}
+		// listeners etc
+
+		// all ok
+		return null;
+	}
+
+	/**
+	 *  Remove a session
+	 *  @return null for success, or error message
+	 */
+	public synchronized String remove(String nick, Properties props) {
+		boolean ok = SAMv3Handler.sSessionsHash.del(nick);
+		SAMMessageSess sess = sessions.remove(nick);
+		if (sess != null) {
+			sess.close();
+			// TODO if 0/0, add back this as listener?
+		} else {
+			ok = false;
+		}
+		if (!ok)
+			return "ID " + nick + " not found";
+		// all ok
+		return null;
+	}
+
+	/**
+	 *  @throws IOException always
+	 */
+	public void receiveDatagramBytes(Destination sender, byte[] data, int proto,
+	                                 int fromPort, int toPort) throws IOException {
+		throw new IOException("master session");
+	}
+
+	/**
+	 *  Does nothing.
+	 */
+	public void stopDatagramReceiving() {}
+	
+	/**
+	 *  @throws IOException always
+	 */
+	public void receiveRawBytes(byte[] data, int proto, int fromPort, int toPort) throws IOException {
+		throw new IOException("master session");
+	}
+
+	/**
+	 *  Does nothing.
+	 */
+	public void stopRawReceiving() {}
+
+
+
+	/////// stream session overrides
+
+	/** @throws I2PException always */
+	@Override
+	public void connect(SAMv3Handler handler, String dest, Properties props) throws I2PException {
+		throw new I2PException("master session");
+	}
+
+	/** @throws SAMException always */
+	@Override
+	public void accept(SAMv3Handler handler, boolean verbose) throws SAMException {
+		throw new SAMException("master session");
+	}
+
+	/** @throws SAMException always */
+	@Override
+	public void startForwardingIncoming(Properties props, boolean sendPorts) throws SAMException {
+		throw new SAMException("master session");
+	}
+
+	/** does nothing */
+	@Override
+	public void stopForwardingIncoming() {}
+
+
+	///// SAMMessageSess interface
+
+	@Override
+	public int getListenProtocol() {
+	    return I2PSession.PROTO_ANY;
+	}
+
+	@Override
+	public int getListenPort() {
+	    return I2PSession.PORT_ANY;
+	}
+
+	/**
+	 * Close the master session
+	 */
+	@Override
+	public void close() {
+		// close sessions?
+		super.close();
+	}
+        
+	// I2PSessionMuxedImpl interface
+
+        public void disconnected(I2PSession session) {
+            if (_log.shouldLog(Log.DEBUG))
+                _log.debug("I2P session disconnected");
+            close();
+        }
+
+        public void errorOccurred(I2PSession session, String message,
+                                  Throwable error) {
+            if (_log.shouldLog(Log.DEBUG))
+                _log.debug("I2P error: " + message, error);
+            close();
+        }
+            
+        public void messageAvailable(I2PSession session, int msgId, long size) {
+            messageAvailable(session, msgId, size, I2PSession.PROTO_UNSPECIFIED,
+                             I2PSession.PORT_UNSPECIFIED, I2PSession.PORT_UNSPECIFIED);
+        }
+
+        /** @since 0.9.24 */
+        public void messageAvailable(I2PSession session, int msgId, long size,
+                                     int proto, int fromPort, int toPort) {
+            try {
+                byte msg[] = session.receiveMessage(msgId);
+                if (msg == null)
+                    return;
+                messageReceived(msg, proto, fromPort, toPort);
+            } catch (I2PSessionException e) {
+                _log.error("Error fetching I2P message", e);
+                close();
+            }
+        }
+        
+        public void reportAbuse(I2PSession session, int severity) {
+            _log.warn("Abuse reported (severity: " + severity + ")");
+            close();
+        }
+
+	private void messageReceived(byte[] msg, int proto, int fromPort, int toPort) {
+		if (_log.shouldWarn())
+			_log.warn("Unhandled message received, length = " + msg.length +
+				" protocol: " + proto + " from port: " + fromPort + " to port: " + toPort);
+	}
+
+	private class StreamAcceptor implements Runnable {
+		
+		public StreamAcceptor() {
+		}
+		
+		public void run() {
+			while (getSocketServer()!=null) {
+				
+				// wait and accept a connection from I2P side
+				I2PSocket i2ps;
+				try {
+					i2ps = getSocketServer().accept();
+					if (i2ps == null)
+						continue;
+				} catch (SocketTimeoutException ste) {
+					continue;
+				} catch (ConnectException ce) {
+					if (_log.shouldLog(Log.WARN))
+						_log.warn("Error accepting", ce);
+					try { Thread.sleep(50); } catch (InterruptedException ie) {}
+					continue;
+				} catch (I2PException ipe) {
+					if (_log.shouldLog(Log.WARN))
+						_log.warn("Error accepting", ipe);
+					break;
+				}
+				int port = i2ps.getLocalPort();
+				SAMMessageSess foundSess = null;
+				Collection<SAMMessageSess> all = sessions.values();
+				for (Iterator<SAMMessageSess> iter = all.iterator(); iter.hasNext(); ) {
+					SAMMessageSess sess = iter.next();
+					if (sess.getListenProtocol() != I2PSession.PROTO_STREAMING) {
+						// remove as we may be going around again below
+						iter.remove();
+						continue;
+					}
+					if (sess.getListenPort() == port) {
+						foundSess = sess;
+						break;
+					}
+				}
+				// We never send streaming out as a raw packet to a default listener,
+				// and we don't allow raw to listen on streaming protocol,
+				// so we don't have to look for a default protocol,
+				// but we do have to look for a default port listener.
+				if (foundSess == null) {
+					for (SAMMessageSess sess : all) {
+						if (sess.getListenPort() == 0) {
+							foundSess = sess;
+							break;
+						}
+					}
+				}
+				if (foundSess != null) {
+					SAMv3StreamSession ssess = (SAMv3StreamSession) foundSess;
+					boolean ok = ssess.queueSocket(i2ps);
+					if (!ok) {
+						_log.logAlways(Log.WARN, "Accept queue overflow for " + ssess);
+						try { i2ps.close(); } catch (IOException ioe) {}
+					}
+				} else {
+					if (_log.shouldLog(Log.WARN))
+						_log.warn("No subsession found for incoming streaming connection on port " + port);
+				}
+			}
+		}
+	}
+}

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

@@ -45,7 +45,7 @@ class SAMDatagramSession extends SAMMessageSession {
      * @throws DataFormatException
      * @throws I2PSessionException 
      */
-    public SAMDatagramSession(String dest, Properties props,
+    protected SAMDatagramSession(String dest, Properties props,
                               SAMDatagramReceiver recv) throws IOException, 
                               DataFormatException, I2PSessionException {
         super(dest, props);
@@ -73,6 +73,20 @@ class SAMDatagramSession extends SAMMessageSession {
         dgramMaker = new I2PDatagramMaker(getI2PSession());
     }
 
+    /**
+     * Create a new SAM DATAGRAM session on an existing I2P session.
+     *
+     * @since 0.9.25
+     */
+    protected SAMDatagramSession(I2PSession sess, int listenPort,
+                              SAMDatagramReceiver recv) throws IOException, 
+                              DataFormatException, I2PSessionException {
+        super(sess, I2PSession.PROTO_DATAGRAM, listenPort);
+
+        this.recv = recv;
+        dgramMaker = new I2PDatagramMaker(getI2PSession());
+    }
+
     /**
      * Send bytes through a SAM DATAGRAM session.
      *

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

@@ -0,0 +1,46 @@
+package net.i2p.sam;
+
+import java.io.Closeable;
+
+import net.i2p.client.I2PSessionException;
+import net.i2p.data.DataFormatException;
+import net.i2p.data.Destination;
+
+/**
+ * Base interface for SAMMessageSession, which is the base for
+ * v1/v3 datagram and raw sessions.
+ * Also implemented by SAMStreamSession.
+ *
+ * @since 0.9.25 pulled from SAMMessageSession
+ */
+interface SAMMessageSess extends Closeable {
+
+    /**
+     * Close a SAM message-based session.
+     */
+    public void close();
+
+    /**
+     * Get the SAM message-based session Destination.
+     *
+     * @return The SAM message-based session Destination.
+     */
+    public Destination getDestination();
+
+    /**
+     * Send bytes through a SAM message-based session.
+     *
+     * @param dest Destination
+     * @param data Bytes to be sent
+     *
+     * @return True if the data was sent, false otherwise
+     * @throws DataFormatException on unknown / bad dest
+     * @throws I2PSessionException on serious error, probably session closed
+     */
+    public boolean sendBytes(String dest, byte[] data, int proto,
+                             int fromPort, int toPort) throws DataFormatException, I2PSessionException;
+
+    public int getListenProtocol();
+
+    public int getListenPort();
+}

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

@@ -33,11 +33,13 @@ import net.i2p.util.Log;
  *
  * @author human
  */
-abstract class SAMMessageSession implements Closeable {
+abstract class SAMMessageSession implements SAMMessageSess {
 
     protected final Log _log;
     private final I2PSession session;
     private final SAMMessageSessionHandler handler;
+    private final int listenProtocol;
+    private final int listenPort;
 
     /**
      * Initialize a new SAM message-based session.
@@ -68,6 +70,33 @@ abstract class SAMMessageSession implements Closeable {
 
         handler = new SAMMessageSessionHandler(destStream, props);
         session = handler.getSession();
+        listenProtocol = I2PSession.PROTO_ANY;
+        listenPort = I2PSession.PORT_ANY;
+        // FIXME don't start threads in constructors
+        Thread t = new I2PAppThread(handler, "SAMMessageSessionHandler");
+        t.start();
+    }
+
+    /**
+     * Initialize a new SAM message-based session using an existing I2PSession.
+     *
+     * @param destStream Input stream containing the destination and private keys (same format as PrivateKeyFile)
+     * @param props Properties to setup the I2P session
+     * @throws IOException
+     * @throws DataFormatException
+     * @throws I2PSessionException 
+     * @since 0.9.25
+     */
+    protected SAMMessageSession(I2PSession sess, int listenProtocol, int listenPort)
+                            throws IOException, DataFormatException, I2PSessionException {
+        _log = I2PAppContext.getGlobalContext().logManager().getLog(getClass());
+        if (_log.shouldLog(Log.DEBUG))
+            _log.debug("Initializing SAM message-based session");
+
+        session = sess;
+        handler = new SAMMessageSessionHandler(session);
+        this.listenProtocol = listenProtocol;
+        this.listenPort = listenPort;
         // FIXME don't start threads in constructors
         Thread t = new I2PAppThread(handler, "SAMMessageSessionHandler");
         t.start();
@@ -82,6 +111,20 @@ abstract class SAMMessageSession implements Closeable {
         return session.getMyDestination();
     }
 
+    /**
+     * @since 0.9.25
+     */
+    public int getListenProtocol() {
+        return listenProtocol;
+    }
+
+    /**
+     * @since 0.9.25
+     */
+    public int getListenPort() {
+        return listenPort;
+    }
+
     /**
      * Send bytes through a SAM message-based session.
      *
@@ -188,7 +231,7 @@ abstract class SAMMessageSession implements Closeable {
      *
      * @author human
      */
-    class SAMMessageSessionHandler implements Runnable, I2PSessionMuxedListener {
+    private class SAMMessageSessionHandler implements Runnable, I2PSessionMuxedListener {
 
         private final I2PSession _session;
         private final Object runningLock = new Object();
@@ -198,8 +241,8 @@ abstract class SAMMessageSession implements Closeable {
          * Create a new SAM message-based session handler
          *
          * @param destStream Input stream containing the destination keys
-	 * @param props Properties to setup the I2P session
+         * @param props Properties to setup the I2P session
-	 * @throws I2PSessionException 
+         * @throws I2PSessionException 
          */
         public SAMMessageSessionHandler(InputStream destStream, Properties props) throws I2PSessionException {
             if (_log.shouldLog(Log.DEBUG))
@@ -218,7 +261,17 @@ abstract class SAMMessageSession implements Closeable {
             if (_log.shouldLog(Log.DEBUG))
                 _log.debug("I2P session connected");
 
-            _session.addMuxedSessionListener(this, I2PSession.PROTO_ANY, I2PSession.PORT_ANY);
+            _session.addMuxedSessionListener(this, listenProtocol, listenPort);
+        }
+                
+        /**
+         * Create a new SAM message-based session handler on an existing I2PSession
+         *
+         * @since 0.9.25
+         */
+        public SAMMessageSessionHandler(I2PSession sess) throws I2PSessionException {
+            _session = sess;
+            _session.addMuxedSessionListener(this, listenProtocol, listenPort);
         }
 
         /**
@@ -257,7 +310,7 @@ abstract class SAMMessageSession implements Closeable {
                 _log.debug("Shutting down SAM message-based session handler");
             
             shutDown();
-            session.removeListener(I2PSession.PROTO_ANY, I2PSession.PORT_ANY);
+            session.removeListener(listenProtocol, listenPort);
             
             try {
                 if (_log.shouldLog(Log.DEBUG))

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

@@ -39,7 +39,7 @@ class SAMRawSession extends SAMMessageSession {
      * @throws DataFormatException
      * @throws I2PSessionException 
      */
-    public SAMRawSession(String dest, Properties props,
+    protected SAMRawSession(String dest, Properties props,
                          SAMRawReceiver recv) throws IOException, DataFormatException, I2PSessionException {
         super(dest, props);
 
@@ -63,6 +63,18 @@ class SAMRawSession extends SAMMessageSession {
         this.recv = recv;
     }
 
+    /**
+     * Create a new SAM RAW session on an existing I2P session.
+     *
+     * @since 0.9.25
+     */
+    protected SAMRawSession(I2PSession sess, int listenProtocol, int listenPort,
+                            SAMRawReceiver recv) throws IOException, 
+                              DataFormatException, I2PSessionException {
+        super(sess, listenProtocol, listenPort);
+        this.recv = recv;
+    }
+
     /**
      * Send bytes through a SAM RAW session.
      *

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

@@ -28,6 +28,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 import net.i2p.I2PAppContext;
 import net.i2p.I2PException;
 import net.i2p.client.I2PClient;
+import net.i2p.client.I2PSession;
 import net.i2p.client.streaming.I2PServerSocket;
 import net.i2p.client.streaming.I2PSocket;
 import net.i2p.client.streaming.I2PSocketManager;
@@ -47,7 +48,7 @@ import net.i2p.util.Log;
  *
  * @author human
  */
-class SAMStreamSession {
+class SAMStreamSession implements SAMMessageSess {
 
     protected final Log _log;
 
@@ -68,6 +69,8 @@ class SAMStreamSession {
 
     // Can we create outgoing connections?
     protected final boolean canCreate;
+    private final int listenProtocol;
+    private final int listenPort;
 
     /** 
      * should we flush every time we get a STREAM SEND, or leave that up to
@@ -105,8 +108,8 @@ class SAMStreamSession {
      * @throws DataFormatException
      * @throws SAMException 
      */
-    public SAMStreamSession(InputStream destStream, String dir,
+    protected SAMStreamSession(InputStream destStream, String dir,
-                            Properties props,  SAMStreamReceiver recv) throws IOException, DataFormatException, SAMException {
+                               Properties props,  SAMStreamReceiver recv) throws IOException, DataFormatException, SAMException {
         this.recv = recv;
         _log = I2PAppContext.getGlobalContext().logManager().getLog(getClass());
 
@@ -170,8 +173,15 @@ class SAMStreamSession {
 
         forceFlush = Boolean.parseBoolean(allprops.getProperty(PROP_FORCE_FLUSH, DEFAULT_FORCE_FLUSH));
         
+        if (Boolean.parseBoolean(props.getProperty("i2p.streaming.enforceProtocol")))
+            listenProtocol = I2PSession.PROTO_STREAMING;
+        else
+            listenProtocol = I2PSession.PROTO_ANY;
+        listenPort = I2PSession.PORT_ANY;
+
 
         if (startAcceptor) {
+            // FIXME don't start threads in constructors
             server = new SAMStreamSessionServer();
             Thread t = new I2PAppThread(server, "SAMStreamSessionServer");
 
@@ -181,6 +191,48 @@ class SAMStreamSession {
         }
     }
 
+    /**
+     * Create a new SAM STREAM session on an existing socket manager.
+     * v3 only.
+     *
+     * @param props Properties to setup the I2P session
+     * @param recv Object that will receive incoming data
+     * @throws IOException
+     * @throws DataFormatException
+     * @throws SAMException 
+     * @since 0.9.25
+     */
+    protected SAMStreamSession(I2PSocketManager mgr, Properties props, SAMStreamReceiver recv, int listenport)
+                               throws IOException, DataFormatException, SAMException {
+        this.recv = recv;
+        _log = I2PAppContext.getGlobalContext().logManager().getLog(getClass());
+        if (_log.shouldLog(Log.DEBUG))
+            _log.debug("SAM STREAM session instantiated");
+        canCreate = true;
+        Properties allprops = (Properties) System.getProperties().clone();
+        allprops.putAll(props);
+        socketMgr = mgr;
+        socketMgr.addDisconnectListener(new DisconnectListener());
+        forceFlush = Boolean.parseBoolean(allprops.getProperty(PROP_FORCE_FLUSH, DEFAULT_FORCE_FLUSH));
+        listenProtocol = I2PSession.PROTO_STREAMING;
+        listenPort = listenport;
+        server = null;
+    }
+
+    /*
+     * @since 0.9.25
+     */
+    public int getListenProtocol() {
+        return listenProtocol;
+    }
+
+    /*
+     * @since 0.9.25
+     */
+    public int getListenPort() {
+        return listenPort;
+    }
+    
     protected class DisconnectListener implements I2PSocketManager.DisconnectListener {
         public void sessionDisconnected() {
             close();
@@ -304,6 +356,15 @@ class SAMStreamSession {
         return true;
     }
 
+    /**
+     *  Unsupported
+     *  @throws DataFormatException always
+     *  @since 0.9.25 moved from subclass SAMv3StreamSession to implement SAMMessageSess
+     */
+    public boolean sendBytes(String s, byte[] b, int pr, int fp, int tp) throws DataFormatException {
+    	throw new DataFormatException(null);
+    }
+
     /** 
      * Create a new SAM STREAM session socket handler, detaching its thread.
      *

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

@@ -40,13 +40,13 @@ import net.i2p.util.Log;
  */
 class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatagramReceiver, SAMStreamReceiver {
     
-    protected SAMRawSession rawSession;
+    protected SAMMessageSess rawSession;
-    protected SAMDatagramSession datagramSession;
+    protected SAMMessageSess datagramSession;
     protected SAMStreamSession streamSession;
 
-    protected SAMRawSession getRawSession() {return rawSession ;}
+    protected final SAMMessageSess getRawSession() { return rawSession; }
-    protected SAMDatagramSession getDatagramSession() {return datagramSession ;}	
+    protected final SAMMessageSess getDatagramSession() { return datagramSession; }	
-    protected SAMStreamSession getStreamSession() {return streamSession ;}
+    protected final SAMStreamSession getStreamSession() { return streamSession; }
 
     protected final long _id;
     private static final AtomicLong __id = new AtomicLong();
@@ -199,14 +199,14 @@ class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatagramRece
                 if (_log.shouldWarn())
                     _log.warn("Error closing socket", e);
             }
-            if (getRawSession() != null) {
+            if (rawSession != null) {
-            	getRawSession().close();
+            	rawSession.close();
             }
-            if (getDatagramSession() != null) {
+            if (datagramSession != null) {
-            	getDatagramSession().close();
+            	datagramSession.close();
             }
-            if (getStreamSession() != null) {
+            if (streamSession != null) {
-            	getStreamSession().close();
+            	streamSession.close();
             }
         }
     }
@@ -218,8 +218,8 @@ class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatagramRece
 
         try{
             if (opcode.equals("CREATE")) {
-                if ((getRawSession() != null) || (getDatagramSession() != null)
+                if ((rawSession != null) || (datagramSession != null)
-                    || (getStreamSession() != null)) {
+                    || (streamSession != null)) {
                     if (_log.shouldLog(Log.DEBUG))
                         _log.debug("Trying to create a session, but one still exists");
                     return writeString("SESSION STATUS RESULT=I2P_ERROR MESSAGE=\"Session already exists\"\n");
@@ -374,12 +374,12 @@ class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatagramRece
 
             Destination dest = null ;
             if (name.equals("ME")) {
-                if (getRawSession() != null) {
+                if (rawSession != null) {
-                    dest = getRawSession().getDestination();
+                    dest = rawSession.getDestination();
-                } else if (getStreamSession() != null) {
+                } else if (streamSession != null) {
-                    dest = getStreamSession().getDestination();
+                    dest = streamSession.getDestination();
-                } else if (getDatagramSession() != null) {
+                } else if (datagramSession != null) {
-                    dest = getDatagramSession().getDestination();
+                    dest = datagramSession.getDestination();
                 } else {
                     if (_log.shouldLog(Log.DEBUG))
                         _log.debug("Lookup for SESSION destination, but session is null");
@@ -411,7 +411,7 @@ class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatagramRece
 
     /* Parse and execute a DATAGRAM message */
     protected boolean execDatagramMessage(String opcode, Properties props) {
-        if (getDatagramSession() == null) {
+        if (datagramSession == null) {
             _log.error("DATAGRAM message received, but no DATAGRAM session exists");
             return false;
         }
@@ -478,7 +478,7 @@ class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatagramRece
 
                 in.readFully(data);
 
-                if (!getDatagramSession().sendBytes(dest, data, proto, fromPort, toPort)) {
+                if (!datagramSession.sendBytes(dest, data, proto, fromPort, toPort)) {
                     _log.error("DATAGRAM SEND failed");
                     // a message send failure is no reason to drop the SAM session
                     // for raw and repliable datagrams, just carry on our merry way
@@ -515,7 +515,7 @@ class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatagramRece
 
     /* Parse and execute a RAW message */
     protected boolean execRawMessage(String opcode, Properties props) {
-        if (getRawSession() == null) {
+        if (rawSession == null) {
             _log.error("RAW message received, but no RAW session exists");
             return false;
         }
@@ -591,7 +591,7 @@ class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatagramRece
 
                 in.readFully(data);
 
-                if (!getRawSession().sendBytes(dest, data, proto, fromPort, toPort)) {
+                if (!rawSession.sendBytes(dest, data, proto, fromPort, toPort)) {
                     _log.error("RAW SEND failed");
                     // a message send failure is no reason to drop the SAM session
                     // for raw and repliable datagrams, just carry on our merry way
@@ -628,7 +628,7 @@ class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatagramRece
 
     /* Parse and execute a STREAM message */
     protected boolean execStreamMessage(String opcode, Properties props) {
-        if (getStreamSession() == null) {
+        if (streamSession == null) {
             _log.error("STREAM message received, but no STREAM session exists");
             return false;
         }
@@ -695,13 +695,13 @@ class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatagramRece
         }
 
         try {
-            if (!getStreamSession().sendBytes(id, getClientSocket().socket().getInputStream(), size)) { // data)) {
+            if (!streamSession.sendBytes(id, getClientSocket().socket().getInputStream(), size)) { // data)) {
                 if (_log.shouldLog(Log.WARN))
                     _log.warn("STREAM SEND [" + size + "] failed");
                 // a message send failure is no reason to drop the SAM session
                 // for style=stream, tell the client the stream failed, and kill the virtual connection..
                 boolean rv = writeString("STREAM CLOSED RESULT=CANT_REACH_PEER ID=" + id + " MESSAGE=\"Send of " + size + " bytes failed\"\n");
-                getStreamSession().closeConnection(id);
+                streamSession.closeConnection(id);
                 return rv;
             }
 
@@ -756,7 +756,7 @@ class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatagramRece
 
         try {
             try {
-                if (!getStreamSession().connect(id, dest, props)) {
+                if (!streamSession.connect(id, dest, props)) {
                     if (_log.shouldLog(Log.DEBUG))
                         _log.debug("STREAM connection failed");
                     return false;
@@ -817,7 +817,7 @@ class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatagramRece
             }
         }
 
-        boolean closed = getStreamSession().closeConnection(id);
+        boolean closed = streamSession.closeConnection(id);
         if ( (!closed) && (_log.shouldLog(Log.WARN)) )
             _log.warn("Stream unable to be closed, but this is non fatal");
         return true;
@@ -835,7 +835,7 @@ class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatagramRece
     
     // SAMRawReceiver implementation
     public void receiveRawBytes(byte data[], int proto, int fromPort, int toPort) throws IOException {
-        if (getRawSession() == null) {
+        if (rawSession == null) {
             _log.error("BUG! Received raw bytes, but session is null!");
             return;
         }
@@ -861,7 +861,7 @@ class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatagramRece
         if (_log.shouldLog(Log.DEBUG))
             _log.debug("stopRawReceiving() invoked");
 
-        if (getRawSession() == null) {
+        if (rawSession == null) {
             _log.error("BUG! Got raw receiving stop, but session is null!");
             return;
         }
@@ -877,7 +877,7 @@ class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatagramRece
     // SAMDatagramReceiver implementation
     public void receiveDatagramBytes(Destination sender, byte data[], int proto,
                                      int fromPort, int toPort) throws IOException {
-        if (getDatagramSession() == null) {
+        if (datagramSession == null) {
             _log.error("BUG! Received datagram bytes, but session is null!");
             return;
         }
@@ -904,7 +904,7 @@ class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatagramRece
         if (_log.shouldLog(Log.DEBUG))
             _log.debug("stopDatagramReceiving() invoked");
 
-        if (getDatagramSession() == null) {
+        if (datagramSession == null) {
             _log.error("BUG! Got datagram receiving stop, but session is null!");
             return;
         }
@@ -921,7 +921,7 @@ class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatagramRece
 
     public void streamSendAnswer( int id, String result, String bufferState ) throws IOException
     {
-        if ( getStreamSession() == null )
+        if ( streamSession == null )
         {
             _log.error ( "BUG! Want to answer to stream SEND, but session is null!" );
             return;
@@ -939,7 +939,7 @@ class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatagramRece
 
     public void notifyStreamSendBufferFree( int id ) throws IOException
     {
-        if ( getStreamSession() == null )
+        if ( streamSession == null )
         {
             _log.error ( "BUG! Stream outgoing buffer is free, but session is null!" );
             return;
@@ -953,7 +953,7 @@ class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatagramRece
 
 
     public void notifyStreamIncomingConnection(int id, Destination d) throws IOException {
-        if (getStreamSession() == null) {
+        if (streamSession == null) {
             _log.error("BUG! Received stream connection, but session is null!");
             return;
         }
@@ -968,7 +968,7 @@ class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatagramRece
     /** @param msg may be null */
     public void notifyStreamOutgoingConnection ( int id, String result, String msg ) throws IOException
     {
-        if ( getStreamSession() == null )
+        if ( streamSession == null )
         {
             _log.error ( "BUG! Received stream connection, but session is null!" );
             return;
@@ -1011,7 +1011,7 @@ class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatagramRece
     }
   
     public void receiveStreamBytes(int id, ByteBuffer data) throws IOException {
-        if (getStreamSession() == null) {
+        if (streamSession == null) {
             _log.error("Received stream bytes, but session is null!");
             return;
         }
@@ -1032,7 +1032,7 @@ class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatagramRece
 
     /** @param msg may be null */
     public void notifyStreamDisconnection(int id, String result, String msg) throws IOException {
-        if (getStreamSession() == null) {
+        if (streamSession == null) {
             _log.error("BUG! Received stream disconnection, but session is null!");
             return;
         }
@@ -1047,7 +1047,7 @@ class SAMv1Handler extends SAMHandler implements SAMRawReceiver, SAMDatagramRece
         if (_log.shouldLog(Log.DEBUG))
             _log.debug("stopStreamReceiving() invoked", new Exception("stopped"));
 
-        if (getStreamSession() == null) {
+        if (streamSession == null) {
             _log.error("BUG! Got stream receiving stop, but session is null!");
             return;
         }

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

@@ -6,17 +6,18 @@
 package net.i2p.sam;
 
 import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress ;
+import java.nio.ByteBuffer;
 import java.util.Properties;
 
+import net.i2p.client.I2PSession;
 import net.i2p.client.I2PSessionException;
 import net.i2p.data.DataFormatException;
 import net.i2p.data.DataHelper;
 import net.i2p.data.Destination;
 import net.i2p.util.Log;
 
-import java.net.InetSocketAddress;
-import java.net.SocketAddress ;
-import java.nio.ByteBuffer;
 
 class SAMv3DatagramSession extends SAMDatagramSession implements Session, SAMDatagramReceiver {
 	
@@ -53,21 +54,28 @@ class SAMv3DatagramSession extends SAMDatagramSession implements Session, SAMDat
 		this.handler = rec.getHandler();
 		
 		Properties props = rec.getProps();
-		String portStr = props.getProperty("PORT");
+		clientAddress = SAMv3RawSession.getSocketAddress(props, handler);
-		if (portStr == null) {
+	}
-			if (_log.shouldDebug())
+
-				_log.debug("receiver port not specified. Current socket will be used.");
+	/**
-			this.clientAddress = null;
+	 *   Build a Datagram Session on an existing i2p session
-		} else {
+	 *   registered with the given nickname
-			int port = Integer.parseInt(portStr);
+	 *   
-			String host = props.getProperty("HOST");
+	 * @param nick nickname of the session
-			if (host == null) {    		
+	 * @throws IOException
-				host = rec.getHandler().getClientIP();
+	 * @throws DataFormatException
-				if (_log.shouldDebug())
+	 * @throws I2PSessionException
-					_log.debug("no host specified. Taken from the client socket : " + host+':'+port);
+	 * @since 0.9.25
-			}
+	 */
-			this.clientAddress = new InetSocketAddress(host, port);
+	public SAMv3DatagramSession(String nick, Properties props, SAMv3Handler handler, I2PSession isess,
-		}
+	                            int listenPort, SAMv3DatagramServer dgServer) 
+			throws IOException, DataFormatException, I2PSessionException {
+		super(isess, listenPort, null);  // to be replaced by this
+		this.nick = nick ;
+		this.recv = this ;  // replacement
+		this.server = dgServer;
+		this.handler = handler;
+		clientAddress = SAMv3RawSession.getSocketAddress(props, handler);
 	}
 
 	public void receiveDatagramBytes(Destination sender, byte[] data, int proto,

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

@@ -48,6 +48,7 @@ class SAMv3Handler extends SAMv1Handler
 {
 	
 	private Session session;
+        // TODO remove singleton, hang off SAMBridge like dgserver
 	public static final SessionsDB sSessionsHash = new SessionsDB();
 	private volatile boolean stolenSocket;
 	private volatile boolean streamForwardingSocket;
@@ -369,9 +370,16 @@ class SAMv3Handler extends SAMv1Handler
 	protected boolean execSessionMessage(String opcode, Properties props) {
 
 		String dest = "BUG!";
-		String nick =  null ;
 		boolean ok = false ;
 
+		String nick = (String) props.remove("ID");
+		if (nick == null)
+			return writeString("SESSION STATUS RESULT=I2P_ERROR MESSAGE=\"ID not specified\"\n");
+
+		String style = (String) props.remove("STYLE");
+		if (style == null && !opcode.equals("REMOVE"))
+			return writeString("SESSION STATUS RESULT=I2P_ERROR MESSAGE=\"No SESSION STYLE specified\"\n");
+
 		try{
 			if (opcode.equals("CREATE")) {
 				if ((this.getRawSession()!= null) || (this.getDatagramSession() != null)
@@ -418,22 +426,6 @@ class SAMv3Handler extends SAMv1Handler
 						return writeString("SESSION STATUS RESULT=INVALID_KEY\n");
 				}
 
-
-				nick = (String) props.remove("ID");
-				if (nick == null) {
-					if (_log.shouldLog(Log.DEBUG))
-						_log.debug("SESSION ID parameter not specified");
-					return writeString("SESSION STATUS RESULT=I2P_ERROR MESSAGE=\"ID not specified\"\n");
-				}
-
-
-				String style = (String) props.remove("STYLE");
-				if (style == null) {
-					if (_log.shouldLog(Log.DEBUG))
-						_log.debug("SESSION STYLE parameter not specified");
-					return writeString("SESSION STATUS RESULT=I2P_ERROR MESSAGE=\"No SESSION STYLE specified\"\n");
-				}
-
 				// Unconditionally override what the client may have set
 				// (iMule sets BestEffort) as None is more efficient
 				// and the client has no way to access delivery notifications
@@ -472,6 +464,13 @@ class SAMv3Handler extends SAMv1Handler
 					SAMv3StreamSession v3 = newSAMStreamSession(nick);
 					streamSession = v3;
 					this.session = v3;
+				} else if (style.equals("MASTER")) {
+					SAMv3DatagramServer dgs = bridge.getV3DatagramServer(props);
+					MasterSession v3 = new MasterSession(nick, dgs, this, allProps);
+					streamSession = v3;
+					datagramSession = v3;
+                                        rawSession = v3;
+					this.session = v3;
 				} else {
 					if (_log.shouldLog(Log.DEBUG))
 						_log.debug("Unrecognized SESSION STYLE: \"" + style +"\"");
@@ -480,6 +479,22 @@ class SAMv3Handler extends SAMv1Handler
 				ok = true ;
 				return writeString("SESSION STATUS RESULT=OK DESTINATION="
 						+ dest + "\n");
+			} else if (opcode.equals("ADD") || opcode.equals("REMOVE")) {
+                                // prevent trouble in finally block
+				ok = true;
+				if (streamSession != null || datagramSession != null || rawSession != null)
+					return writeString("SESSION STATUS RESULT=I2P_ERROR MESSAGE=\"Not a MASTER session\"\n");
+				MasterSession msess = (MasterSession) session;
+				String msg;
+				if (opcode.equals("ADD")) {
+					msg = msess.add(nick, style, props);
+				} else {
+					msg = msess.remove(nick, props);
+				}
+				if (msg == null)
+					return writeString("SESSION STATUS RESULT=OK MESSAGE=\"" + opcode + ' ' + nick + "\"\n");
+				else
+					return writeString("SESSION STATUS RESULT=I2P_ERROR MESSAGE=\"" + msg + "\"\n");
 			} else {
 				if (_log.shouldLog(Log.DEBUG))
 					_log.debug("Unrecognized SESSION message opcode: \""

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

@@ -10,6 +10,7 @@ import java.net.SocketAddress;
 import java.nio.ByteBuffer;
 import java.util.Properties;
 
+import net.i2p.client.I2PSession;
 import net.i2p.client.I2PSessionException;
 import net.i2p.data.DataFormatException;
 import net.i2p.data.DataHelper;
@@ -19,7 +20,7 @@ import net.i2p.util.Log;
  * @author MKVore
  *
  */
-class SAMv3RawSession extends SAMRawSession  implements Session, SAMRawReceiver {
+class SAMv3RawSession extends SAMRawSession implements Session, SAMRawReceiver {
 	
 	private final String nick;
 	private final SAMv3Handler handler;
@@ -42,34 +43,60 @@ class SAMv3RawSession extends SAMRawSession  implements Session, SAMRawReceiver
 			throws IOException, DataFormatException, I2PSessionException {
 		super(SAMv3Handler.sSessionsHash.get(nick).getDest(),
 		      SAMv3Handler.sSessionsHash.get(nick).getProps(),
-		      SAMv3Handler.sSessionsHash.get(nick).getHandler()  // to be replaced by this
+		      null  // to be replaced by this
 		);
 		this.nick = nick ;
 		this.recv = this ;  // replacement
 		this.server = dgServer;
-
 		SessionRecord rec = SAMv3Handler.sSessionsHash.get(nick);
 		if (rec == null)
 			throw new InterruptedIOException() ;
 		this.handler = rec.getHandler();
 		Properties props = rec.getProps();
+		clientAddress = getSocketAddress(props, handler);
+		_sendHeader = ((handler.verMajor == 3 && handler.verMinor >= 2) || handler.verMajor > 3) &&
+		              Boolean.parseBoolean(props.getProperty("HEADER"));
+	}
+
+	/**
+	 *   Build a Raw Session on an existing i2p session
+	 *   registered with the given nickname
+	 *   
+	 * @param nick nickname of the session
+	 * @throws IOException
+	 * @throws DataFormatException
+	 * @throws I2PSessionException
+	 * @since 0.9.25
+	 */
+	public SAMv3RawSession(String nick, Properties props, SAMv3Handler handler, I2PSession isess,
+	                       int listenProtocol, int listenPort, SAMv3DatagramServer dgServer) 
+			throws IOException, DataFormatException, I2PSessionException {
+		super(isess, listenProtocol, listenPort, null);  // to be replace by this
+		this.nick = nick ;
+		this.recv = this ;  // replacement
+		this.server = dgServer;
+		this.handler = handler;
+		clientAddress = getSocketAddress(props, handler);
+		_sendHeader = ((handler.verMajor == 3 && handler.verMinor >= 2) || handler.verMajor > 3) &&
+		              Boolean.parseBoolean(props.getProperty("HEADER"));
+	}
+	
+	/**
+	 *  @return null if PORT not set
+	 *  @since 0.9.25 moved from constructor
+	 */
+	static SocketAddress getSocketAddress(Properties props, SAMv3Handler handler) {
 		String portStr = props.getProperty("PORT") ;
 		if (portStr == null) {
-			if (_log.shouldLog(Log.DEBUG))
+			return null;
-				_log.debug("receiver port not specified. Current socket will be used.");
-			this.clientAddress = null;
 		} else {
 			int port = Integer.parseInt(portStr);
 			String host = props.getProperty("HOST");
 			if ( host==null ) {
-				host = rec.getHandler().getClientIP();
+				host = handler.getClientIP();
-				if (_log.shouldLog(Log.DEBUG))
-		    			_log.debug("no host specified. Taken from the client socket : " + host +':'+port);
 			}
-			this.clientAddress = new InetSocketAddress(host, port);
+			return new InetSocketAddress(host, port);
 		}
-		_sendHeader = ((handler.verMajor == 3 && handler.verMinor >= 2) || handler.verMajor > 3) &&
-                              Boolean.parseBoolean(props.getProperty("HEADER"));
 	}
 
 	public void receiveRawBytes(byte[] data, int proto, int fromPort, int toPort) throws IOException {

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

@@ -21,6 +21,7 @@ import java.nio.ByteBuffer;
 import java.nio.channels.SocketChannel;
 import java.security.GeneralSecurityException;
 import java.util.Properties;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import javax.net.ssl.SSLException;
@@ -30,6 +31,7 @@ import net.i2p.I2PAppContext;
 import net.i2p.I2PException;
 import net.i2p.client.streaming.I2PServerSocket;
 import net.i2p.client.streaming.I2PSocket;
+import net.i2p.client.streaming.I2PSocketManager;
 import net.i2p.client.streaming.I2PSocketOptions;
 import net.i2p.data.DataFormatException;
 import net.i2p.data.Destination;
@@ -46,13 +48,16 @@ import net.i2p.util.Log;
 class SAMv3StreamSession  extends SAMStreamSession implements Session
 {
 
-		private static final int BUFFER_SIZE = 1024 ;
+		private static final int BUFFER_SIZE = 1024;
+		private static final int MAX_ACCEPT_QUEUE = 64;
 		
 		private final Object socketServerLock = new Object();
 		/** this is ONLY set for FORWARD, not for ACCEPT */
 		private I2PServerSocket socketServer;
 		/** this is the count of active ACCEPT sockets */
 		private final AtomicInteger _acceptors = new AtomicInteger();
+		/** for subsession only, null otherwise */
+		private final LinkedBlockingQueue<I2PSocket> _acceptQueue;
 
 		private static I2PSSLSocketFactory _sslSocketFactory;
 	
@@ -79,6 +84,55 @@ class SAMv3StreamSession  extends SAMStreamSession implements Session
                       getDB().get(login).getProps(),
                       getDB().get(login).getHandler());
 	    	this.nick = login ;
+		_acceptQueue = null;
+	    }
+
+	    /**
+	     *   Build a Datagram Session on an existing I2P session
+	     *   registered with the given nickname
+	     *   
+	     * @param nick nickname of the session
+	     * @throws IOException
+	     * @throws DataFormatException
+	     * @throws I2PSessionException
+	     * @since 0.9.25
+	     */
+	    public SAMv3StreamSession(String login, Properties props, SAMv3Handler handler, I2PSocketManager mgr,
+	                                int listenPort) throws IOException, DataFormatException, SAMException {
+		super(mgr, props, handler, listenPort);
+		this.nick = login ;
+		_acceptQueue = new LinkedBlockingQueue<I2PSocket>(MAX_ACCEPT_QUEUE);
+	    }
+
+	    /**
+	     * Put a socket on the accept queue.
+	     * Only for subsession, throws IllegalStateException otherwise.
+	     *   
+	     * @return success, false if full
+	     * @since 0.9.25
+	     */
+	    public boolean queueSocket(I2PSocket sock) {
+		if (_acceptQueue == null)
+		    throw new IllegalStateException();
+		return _acceptQueue.offer(sock);
+	    }
+
+	    /**
+	     * Take a socket from the accept queue.
+	     * Only for subsession, throws IllegalStateException otherwise.
+	     *   
+	     * @since 0.9.25
+	     */
+	    private I2PSocket acceptSocket() throws ConnectException {
+		if (_acceptQueue == null)
+		    throw new IllegalStateException();
+		try {
+			return _acceptQueue.take();
+		} catch (InterruptedException ie) {
+			ConnectException ce = new ConnectException("interrupted");
+			ce.initCause(ie);
+			throw ce;
+		}
 	    }
 
 	    public static SessionsDB getDB()
@@ -185,10 +239,13 @@ class SAMv3StreamSession  extends SAMStreamSession implements Session
 			}
 		}
 
-		I2PSocket i2ps;
+		I2PSocket i2ps = null;
 		_acceptors.incrementAndGet();
 		try {
-			i2ps = socketMgr.getServerSocket().accept();
+			if (_acceptQueue != null)
+				i2ps = acceptSocket();
+			else
+				i2ps = socketMgr.getServerSocket().accept();
 		} finally {
 			_acceptors.decrementAndGet();
 		}
@@ -257,25 +314,23 @@ class SAMv3StreamSession  extends SAMStreamSession implements Session
 	    		this.socketServer = this.socketMgr.getServerSocket();
 	    	}
 	    	
-	    	SocketForwarder forwarder = new SocketForwarder(host, port, isSSL, this, verbose, sendPorts);
+	    	SocketForwarder forwarder = new SocketForwarder(host, port, isSSL, verbose, sendPorts);
 	    	(new I2PAppThread(rec.getThreadGroup(), forwarder, "SAMV3StreamForwarder")).start();
 	    }
 	    
 	    /**
 	     *  Forward sockets from I2P to the host/port provided
 	     */
-	    private static class SocketForwarder implements Runnable
+	    private class SocketForwarder implements Runnable
 	    {
 	    	private final String host;
 	    	private final int port;
-	    	private final SAMv3StreamSession session;
 	    	private final boolean isSSL, verbose, sendPorts;
 	    	
 	    	SocketForwarder(String host, int port, boolean isSSL,
-		                SAMv3StreamSession session, boolean verbose, boolean sendPorts) {
+		                boolean verbose, boolean sendPorts) {
 	    		this.host = host ;
 	    		this.port = port ;
-	    		this.session = session ;
 	    		this.verbose = verbose ;
 	    		this.sendPorts = sendPorts;
 			this.isSSL = isSSL;
@@ -283,12 +338,15 @@ class SAMv3StreamSession  extends SAMStreamSession implements Session
 	    	
 	    	public void run()
 	    	{
-	    		while (session.getSocketServer()!=null) {
+	    		while (getSocketServer() != null) {
 	    			
 	    			// wait and accept a connection from I2P side
 	    			I2PSocket i2ps;
 	    			try {
-	    				i2ps = session.getSocketServer().accept();
+					if (_acceptQueue != null)
+						i2ps = acceptSocket();
+					else
+		    				i2ps = getSocketServer().accept();
 	    				if (i2ps == null)
 		    				continue;
 				} catch (SocketTimeoutException ste) {
@@ -437,7 +495,7 @@ class SAMv3StreamSession  extends SAMStreamSession implements Session
 		}
 	    }
 	    
-	    private I2PServerSocket getSocketServer()
+	    protected I2PServerSocket getSocketServer()
 	    {
 	    	synchronized ( this.socketServerLock ) {
 	    		return this.socketServer ;
@@ -479,13 +537,4 @@ class SAMv3StreamSession  extends SAMStreamSession implements Session
 	    public void close() {
 	        socketMgr.destroySocketManager();
 	    }
-
-	    /**
-	     *  Unsupported
-	     *  @throws DataFormatException always
-	     */
-	    public boolean sendBytes(String s, byte[] b, int pr, int fp, int tp) throws DataFormatException
-	    {
-	    	throw new DataFormatException(null);
-	    }
 }