Fixed build.xml to detect os, and launch 'jythonc' or 'jythonc.bat'

according to whether we're running on *nix or windoze. build.xml
should now work on your platform, as long as you have jython installed
and jython is on your execution path.

Got SAM STREAMs working - test code added to i2psamclient.py
as function demoSTREAM()

apps/sam/jython/build.xml

@@ -11,17 +11,25 @@
     </target>
 
     <target name="jar">
-        <mkdir dir="./build" />
+
-        <exec executable="jythonc">
+        <condition property="jythonext" value=".bat">
+            <os family="windows" />
+        </condition>
+        <condition property="jythonext" value="">
+            <not>
+                <os family="windows" />
+            </not>
+        </condition>
+
+        <exec executable="jythonc${jythonext}" dir=".">
             <env key="CLASSPATH" path="../../../core/java/build/i2p.jar:../../ministreaming/java/build/mstreaming.jar"/>
-            <arg value="--jar"/>
+            <arg value="--jar"/><arg path="./i2psam.jar"/>
-            <arg value="./build/i2psam.jar"/>
+            <arg path="./src/i2psam.py"/>
-            <arg value="./src/i2psam.py"/>
         </exec>
     </target>
 
     <target name="clean">
-        <delete dir="./build" />
+        <delete file="i2psam.jar" />
         <delete dir="./jpywork" />
     </target>
 

apps/sam/jython/src/i2psam.py

@@ -79,7 +79,7 @@ i2cpPort = 7654
 # logging settings
 
 # 1=v.quiet, 2=normal, 3=verbose, 4=debug, 5=painful
-verbosity = 5
+verbosity = 2
 
 # change to a filename to log there instead
 logfile = sys.stdout
@@ -851,23 +851,28 @@ class I2PSocket:
         self.dest = dest
     
         if kw.has_key('sock') \
-                and kw.has_key('dest') \
                 and kw.has_key('remdest') \
                 and kw.has_key('instream') \
                 and kw.has_key('outstream'):
+    
             # wrapping an accept()'ed connection
+            log(4, "accept()'ed a connection, wrapping...")
+    
             self.sock = kw['sock']
-            self.dest = kw['dest']
+            self.dest = dest
             self.remdest = kw['remdest']
             self.instream = kw['instream']
             self.outstream = kw['outstream']
         else:
+            log(4, "creating new I2PSocket %s" % dest)
+    
             # process keywords
             self.host = kw.get('host', self.host)
             self.port = int(kw.get('port', self.port))
     
             # we need a factory, don't we?
             self.sockmgrFact = i2p.client.streaming.I2PSocketManagerFactory()
+    
     #@-node:__init__
     #@+node:bind
     def bind(self, dest=None):
@@ -882,7 +887,8 @@ class I2PSocket:
             self.dest = dest
         elif not self.dest:
             # create new dest, client should interrogate it at some time
-            self.dest = Destination()
+            log(4, "bind: socket has no dest, creating one")
+            self.dest = I2PDestination()
     #@-node:bind
     #@+node:listen
     def listen(self, *args, **kw):
@@ -894,6 +900,8 @@ class I2PSocket:
             raise I2PSocketError(".sockmgr already present - have you already called listen?")
         if not self.dest:
             raise I2PSocketError("socket is not bound to a destination")
+    
+        log(4, "listening on socket")
         
         # create the socket manager
         self._createSockmgr()
@@ -936,11 +944,12 @@ class I2PSocket:
         to different dests.
         """
         # sanity check
-        #if self.sockmgr:
+        if self.sockmgr:
-        #    raise I2PSocketError(".sockmgr already present - have you already called listen/connect?")
+            raise I2PSocketError(".sockmgr already present - have you already called listen/connect?")
     
         # create whole new dest if none was provided to constructor
         if self.dest is None:
+            log(4, "connect: creating whole new dest")
             self.dest = I2PDestination()
     
         # create the socket manager
@@ -951,6 +960,7 @@ class I2PSocket:
     
         opts = net.i2p.client.streaming.I2PSocketOptions()
         try:
+            log(4, "trying to connect to %s" % remdest.toBase64())
             sock = self.sock = self.sockmgr.connect(remdest._item, opts)
             self.remdest = remdest
         except:
@@ -962,8 +972,8 @@ class I2PSocket:
         sockobj = I2PSocket(dest=self.dest,
                             remdest=remdest,
                             sock=sock,
-                            instream=instream,
+                            instream=self.instream,
-                            outstream=outstream)
+                            outstream=self.outstream)
         self._connected = 1
         return sockobj
     #@-node:connect
@@ -1004,23 +1014,27 @@ class I2PSocket:
             raise I2PSocketError("Socket is not connected")
     
         # and write it out
-        #print "send: writing '%s' to outstream..." % repr(buf)
+        log(4, "send: writing '%s' to outstream..." % repr(buf))
         outstream = self.outstream
         for c in buf:
             outstream.write(ord(c))
     
         # flush just in case
-        #print "send: flushing..."
+        log(4, "send: flushing...")
         self.outstream.flush()
     
-        #print "send: done"
+        log(4, "send: done")
+    
     #@-node:send
     #@+node:available
     def available(self):
         """
         Returns the number of bytes available for recv()
         """
-        return self.sock.available()
+        #print "available: sock is %s" % repr(self.sock)
+    
+        return self.instream.available()
+    
     
     #@-node:available
     #@+node:close
@@ -1049,6 +1063,9 @@ class I2PSocket:
     #@+node:_createSockmgr
     def _createSockmgr(self):
     
+        if getattr(self, 'sockmgr', None):
+            return
+    
         #options = {jI2PClient.PROP_TCP_HOST: self.host,
         #           jI2PClient.PROP_TCP_PORT: self.port}
         options = {}
@@ -1606,6 +1623,26 @@ class I2PSamClientHandler(StreamRequestHandler):
     
             else: # STREAM
                 # no need to create session object, because we're using streaming api
+                log(4, "Creating STREAM session")
+                
+                # what kind of stream?
+                direction = args.get('DIRECTION', 'BOTH')
+                if direction not in ['BOTH', 'RECEIVE', 'CREATE']:
+                    self.samSend("SESSION", "STATUS",
+                        RESULT="I2P_ERROR",
+                        MESSAGE="Illegal_direction_keyword_%s" % direction.replace(" ","_"),
+                        )
+                    return
+    
+                if direction == 'BOTH':
+                    self.canConnect = 1
+                    self.canAccept = 1
+                elif direction == 'RECEIVE':
+                    self.canConnect = 0
+                    self.canAccept = 1
+                elif direction == 'CREATE':
+                    self.canConnect = 1
+                    self.canAccept = 0
     
                 # but we do need to mark it as being in use
                 localsessions[destb64] = globalsessions[destb64] = None
@@ -1613,8 +1650,9 @@ class I2PSamClientHandler(StreamRequestHandler):
                 # make a local socket
                 sock = self.samSock = I2PSocket(dest)
     
-                # and we also need to fire up a socket listener
+                # and we also need to fire up a socket listener, if not CREATE-only
-                thread.start_new_thread(self.threadSocketListener, (sock, dest))
+                if self.canAccept:
+                    thread.start_new_thread(self.threadSocketListener, (sock, dest))
     
             # finally, we can reply with the good news
             self.samSend("SESSION", "STATUS",
@@ -1663,22 +1701,39 @@ class I2PSamClientHandler(StreamRequestHandler):
     
         if subtopic == 'CONNECT':
             # who are we connecting to again?
-            remdest = I2PDestionation(b64=args['DESTINATION'])
+            remdest = I2PDestination(base64=args['DESTINATION'])
-            id = args['ID']
+            id = int(args['ID'])
         
             try:
+                log(4, "Trying to connect to remote peer %s..." % args['DESTINATION'])
                 sock = self.samSock.connect(remdest)
+                log(4, "Connected to remote peer %s..." % args['DESTINATION'])
                 self.localstreams[id] = sock
                 self.samSend("STREAM", "STATUS",
                              RESULT='OK',
                              ID=id,
                              )
+                thread.start_new_thread(self.threadSocketReceiver, (sock, id))
+    
             except:
+                log(4, "Failed to connect to remote peer %s..." % args['DESTINATION'])
                 self.samSend("STREAM", "STATUS",
                              RESULT='I2P_ERROR',
                              MESSAGE='exception_on_connect',
+                             ID=id,
                              )
     
+        elif subtopic == 'SEND':
+            # send to someone
+            id = int(args['ID'])
+            try:
+                sock = self.localstreams[id]
+                sock.send(args['DATA'])
+            except:
+                logException(4, "send failed")
+    
+    
+    
     
     #@-node:on_STREAM
     #@+node:on_DATAGRAM
@@ -1840,18 +1895,24 @@ class I2PSamClientHandler(StreamRequestHandler):
         """
         destb64 = dest.toBase64()
     
-        log(4, "Listening for connections to %s..." % destb64[:40])
+        log(4, "Listening for connections to %s..." % destb64)
     
+        sock.bind()
         sock.listen()
     
         while 1:
+            log(4, "Awaiting next connection to %s..." % destb64)
             newsock = sock.accept()
-            
+            log(4, "Got connection to %s..." % destb64)
+    
             # need an id, negative
             id = - self.server.samAllocId()
     
             # register it in local and global streams
             self.localstreams[id] = self.globalstreams[id] = newsock
+    
+            # fire up the receiver thread
+            thread.start_new_thread(self.threadSocketReceiver, (newsock, id))
             
             # who is connected to us?
             remdest = newsock.remdest
@@ -1861,7 +1922,55 @@ class I2PSamClientHandler(StreamRequestHandler):
             self.samSend("STREAM", "CONNECTED",
                          DESTINATION=remdest_b64,
                          ID=id)
+    
     #@-node:threadSocketListener
+    #@+node:threadSocketReceiver
+    def threadSocketReceiver(self, sock, id):
+        """
+        One of these gets launched each time a new stream connection
+        is created. Due to the lack of callback mechanism within the
+        ministreaming API, we have to actively poll for and send back
+        received data
+        """
+        while 1:
+            #avail = sock.available()
+            #if avail <= 0:
+            #    print "threadSocketReceiver: waiting for data on %s (%s avail)..." % (id, avail)
+            #    time.sleep(5)
+            #    continue
+            #log(4, "reading a byte")
+    
+            try:
+                buf = sock.recv(1)
+            except:
+                logException(4, "Exception reading first byte")
+            
+            if buf == '':
+                log(4, "stream closed")
+    
+                # notify a close
+                self.samSend("STREAM", "CLOSED",
+                             ID=id)
+                return
+    
+            # grab more if there's any available
+            navail = sock.available()
+            if navail > 0:
+                #log(4, "%d more bytes available, reading..." % navail)
+                rest = sock.recv(navail)
+                buf += rest
+            
+            # send if off
+            log(4, "got from peer: %s" % repr(buf))
+            
+            self.samSend("STREAM", "RECEIVED", buf,
+                         ID=id,
+                         )
+    
+    
+    
+    
+    #@-node:threadSocketReceiver
     #@+node:samParse
     def samParse(self, flds):
         """
@@ -1879,7 +1988,7 @@ class I2PSamClientHandler(StreamRequestHandler):
             try:
                 name, val = arg.split("=", 1)
             except:
-                logException(3, "failed to process %s" % repr(arg))
+                logException(3, "failed to process %s in %s" % (repr(arg), repr(flds)))
                 raise
             dargs[name] = val
     
@@ -2108,106 +2217,6 @@ def logException(level, msg=''):
     traceback.print_exc(file=s)
     log(level, "%s\n%s" % (s.getvalue(), msg), 1)
 #@-node:logException
-#@+node:usage
-def usage(detailed=0):
-    
-    print "Usage: %s <options> [<command>]" % sys.argv[0]
-    if not detailed:
-        print "Run with '-h' to get detailed help"
-        sys.exit(0)
-
-    print "I2PSAM is a bridge that allows I2P client programs to access the"
-    print "I2P network by talking over a plaintext socket connection."
-    print "References:"
-    print "   - http://www.freenet.org.nz/i2p - source, doco, downloadables"
-    print "   - http://drupal.i2p.net/node/view/144 - I2P SAM specification"
-    print
-    print "Options:"
-    print "  -h, -?, --help        - display this help"
-    print "  -v, --version         - print program version"
-    print "  -V, --verbosity=n     - set verbosity to n, default 2, 1==quiet, 4==noisy"
-    print "  -H, --listenhost=host - specify host to listen on for client connections"
-    print "  -P, --listenport=port - port to listen on for client connections"
-    print "      --i2cphost=host   - hostname of I2P router's I2CP interface"
-    print "      --i2cpport=port   - port of I2P router's I2CP interface"
-    print
-    print "Commands:"
-    print "     (run with no commands to launch SAM server)"
-    print "     samserver - runs as a SAM server"
-    print "     test - run a suite of self-tests"
-    print
-    
-    sys.exit(0)
-
-
-
-#@-node:usage
-#@+node:main
-def main():
-
-    argv = sys.argv
-    argc = len(argv)
-
-    try:
-        opts, args = getopt.getopt(sys.argv[1:],
-                                   "h?vV:H:P:",
-                                   ['help', 'version', 'verbosity=',
-                                    'listenhost=', 'listenport=',
-                                    'i2cphost=', 'i2cpport=',
-                                    ])
-    except:
-        traceback.print_exc(file=sys.stdout)
-        usage("You entered an invalid option")
-
-    cmd = 'samserver'
-
-    # we prolly should pass all these parms in constructor call, but
-    # what the heck!
-    #global verbosity, i2psamhost, i2psamport, i2cpHost, i2cpPort
-    
-    serveropts = {}
-
-    for opt, val in opts:
-        if opt in ['-h', '-?', '--help']:
-            usage(1)
-        elif opt in ['-v', '--version']:
-            print "I2P SAM version %s" % version
-            sys.exit(0)
-        elif opt in ['-V', '--verbosity']:
-            serveropts['verbosity'] = int(val)
-        elif opt in ['-H', '--listenhost']:
-            serveropts['host'] = val
-        elif opt in ['-P', '--listenport']:
-            serveropts['port'] = int(val)
-        elif opt in ['--i2cphost']:
-            serveropts['i2cphost'] = val
-        elif opt in ['--i2cpport']:
-            serveropts['i2cpport'] = int(val)
-        else:
-            usage(0)
-
-    if len(args) == 0:
-        cmd = 'samserver'
-    else:
-        cmd = args[0]
-
-    if cmd == 'samserver':
-
-        log(2, "Running I2P SAM Server...")
-        server = I2PSamServer(**serveropts)
-        server.run()
-
-    elif cmd == 'test':
-        
-        print "RUNNING I2P Jython TESTS"
-        testsigs()
-        testdests()
-        testsession()
-        testsocket()
-
-    else:
-        usage(0)
-#@-node:main
 #@+node:testdests
 def testdests():
     """
@@ -2356,7 +2365,7 @@ def testsession():
     print "session tests passed!"
 #@-node:testsession
 #@+node:testsocket
-def testsocket():
+def testsocket(bidirectional=0):
 
     global d1, d2, s1, s2
 
@@ -2416,6 +2425,19 @@ def testsocket():
     print "launching server thread..."
     thread.start_new_thread(servThread, (sServer,))
 
+    if bidirectional:
+        # dummy thread which accepts connections TO client socket
+        def threadDummy(s):
+            print "dummy: listening"
+            s.listen()
+            print "dummy: accepting"
+    
+            sock = s.accept()
+            print "dummy: got connection"
+    
+        print "test - launching dummy client accept thread"
+        thread.start_new_thread(threadDummy, (sClient,))
+
     print "client: trying to connect"
     sClient.connect(dServer)
 
@@ -2432,7 +2454,119 @@ def testsocket():
 
     print "I2PSocket test apparently succeeded"
 
+
 #@-node:testsocket
+#@+node:usage
+def usage(detailed=0):
+    
+    print "Usage: %s <options> [<command>]" % sys.argv[0]
+    if not detailed:
+        print "Run with '-h' to get detailed help"
+        sys.exit(0)
+
+    print "I2PSAM is a bridge that allows I2P client programs to access the"
+    print "I2P network by talking over a plaintext socket connection."
+    print "References:"
+    print "   - http://www.freenet.org.nz/i2p - source, doco, downloadables"
+    print "   - http://drupal.i2p.net/node/view/144 - I2P SAM specification"
+    print
+    print "Options:"
+    print "  -h, -?, --help        - display this help"
+    print "  -v, --version         - print program version"
+    print "  -V, --verbosity=n     - set verbosity to n, default 2, 1==quiet, 4==noisy"
+    print "  -H, --listenhost=host - specify host to listen on for client connections"
+    print "  -P, --listenport=port - port to listen on for client connections"
+    print "      --i2cphost=host   - hostname of I2P router's I2CP interface"
+    print "      --i2cpport=port   - port of I2P router's I2CP interface"
+    print
+    print "Commands:"
+    print "     (run with no commands to launch SAM server)"
+    print "     samserver - runs as a SAM server"
+    print "     test - run a suite of self-tests"
+    print "     testsocket - run only the socket test"
+    print "     testbidirsocket - run socket test in bidirectional mode"
+    print
+    
+    sys.exit(0)
+#@-node:usage
+#@+node:main
+def main():
+
+    argv = sys.argv
+    argc = len(argv)
+
+    # -------------------------------------------------
+    # do the getopt command line parsing
+
+    try:
+        opts, args = getopt.getopt(sys.argv[1:],
+                                   "h?vV:H:P:",
+                                   ['help', 'version', 'verbosity=',
+                                    'listenhost=', 'listenport=',
+                                    'i2cphost=', 'i2cpport=',
+                                    ])
+    except:
+        traceback.print_exc(file=sys.stdout)
+        usage("You entered an invalid option")
+
+    #print "args=%s" % args
+
+    serveropts = {}
+    for opt, val in opts:
+        if opt in ['-h', '-?', '--help']:
+            usage(1)
+        elif opt in ['-v', '--version']:
+            print "I2P SAM version %s" % version
+            sys.exit(0)
+        elif opt in ['-V', '--verbosity']:
+            serveropts['verbosity'] = int(val)
+        elif opt in ['-H', '--listenhost']:
+            serveropts['host'] = val
+        elif opt in ['-P', '--listenport']:
+            serveropts['port'] = int(val)
+        elif opt in ['--i2cphost']:
+            serveropts['i2cphost'] = val
+        elif opt in ['--i2cpport']:
+            serveropts['i2cpport'] = int(val)
+        else:
+            usage(0)
+
+    # --------------------------------------------------
+    # now run in required mode, default is 'samserver'
+
+    if len(args) == 0:
+        cmd = 'samserver'
+    else:
+        cmd = args[0]
+
+    if cmd == 'samserver':
+
+        log(2, "Running I2P SAM Server...")
+        server = I2PSamServer(**serveropts)
+        server.run()
+
+    elif cmd == 'test':
+        
+        print "RUNNING full I2PSAM Jython TEST SUITE"
+        testsigs()
+        testdests()
+        testsession()
+        testsocket()
+
+    elif cmd == 'testsocket':
+        
+        print "RUNNING SOCKET TEST"
+        testsocket(0)
+
+    elif cmd == 'testbidirsocket':
+        print "RUNNING BIDIRECTIONAL SOCKET TEST"
+        testsocket(1)
+
+    else:
+        # spit at unrecognised option
+        usage(0)
+
+#@-node:main
 #@+node:MAINLINE
 if __name__ == '__main__':
     main()

apps/sam/python/src/i2psamclient.py

@@ -17,6 +17,8 @@ Run this module without arguments to see a demo in action
 #@+node:imports
 import sys, os, socket, thread, threading, Queue, traceback, StringIO, time
 
+from pdb import set_trace
+
 #@-node:imports
 #@+node:globals
 # -----------------------------------------
@@ -53,6 +55,11 @@ class I2PCommandFail(Exception):
     A failure in an I2CP command
     """
     pass
+
+class I2PStreamClosed(Exception):
+    """
+    Stream is not open
+    """
 #@-node:exceptions
 #@+node:class I2PSamClient
 class I2PSamClient:
@@ -110,10 +117,20 @@ class I2PSamClient:
         self.qSession = Queue.Queue()
         self.qDatagrams = Queue.Queue()
         self.qRawMessages = Queue.Queue()
+    
         self.namingReplies = {}
         self.namingCache = {}
+    
+        self.streams = {} # currently open streams, keyed by id
+        self.streamConnectReplies = {} # holds queues awaiting connect resp, keyed by id
+        self.qNewStreams = Queue.Queue() # incoming connections
+    
+        self.samNextIdLock = threading.Lock()
+        self.samNextId = 1
+    
         self.isRunning = 1
     
+    
         log(4, "trying connection to SAM server...")
         try:
             self.sock.connect((self.host, self.port))
@@ -312,6 +329,8 @@ class I2PSamClient:
             - dest - base64 private destination
         
         Keywords:
+            - direction - only used for STREAM sessions, can be RECEIVE,
+              CREATE or BOTH (default BOTH)
             - i2cphost - hostname for the SAM bridge to contact i2p router on
             - i2cpport - port for the SAM bridge to contact i2p router on
         
@@ -322,6 +341,20 @@ class I2PSamClient:
         kw1 = dict(kw)
         kw1['STYLE'] = self.samStyle = style
         kw1['DESTINATION'] = dest
+        if style == 'STREAM':
+            direction = kw.get('direction', 'BOTH')
+            kw1['DIRECTION'] = direction
+            if direction == 'BOTH':
+                self.canAccept = 1
+                self.canConnect = 1
+            elif direction == 'RECEIVE':
+                self.canAccept = 1
+                self.canConnect = 0
+            elif direction == 'CREATE':
+                self.canAccept = 0
+                self.canConnect = 1
+            else:
+                raise I2PCommandFail("direction keyword must be one of RECEIVE, CREATE or BOTH")
     
         # stick in i2cp host/port if specified
         if kw.has_key('i2cphost'):
@@ -423,6 +456,92 @@ class I2PSamClient:
                 return None
         return self.qDatagrams.get()
     #@-node:samDatagramReceive
+    #@+node:samStreamConnect
+    def samStreamConnect(self, dest):
+        """
+        Makes a STREAM connection to a remote dest
+    
+        STREAM STATUS
+        RESULT=$result
+        ID=$id
+        [MESSAGE=...]
+        """
+        # need an ID
+        id = self.samAllocId()
+        
+        # create queue for connect reply
+        q = self.streamConnectReplies[id] = Queue.Queue()
+    
+        # send req
+        self.samSend("STREAM", "CONNECT",
+                     ID=id,
+                     DESTINATION=dest,
+                     )
+    
+        # await reply - comes back as a dict
+        resp = q.get()
+    
+        # ditch queue
+        del self.streamConnectReplies[id]
+        del q
+        
+        # check out response
+        result = resp['RESULT']
+        if result == 'OK':
+            conn = I2PSAMStream(self, id, dest)
+            self.streams[id] = conn
+            return conn
+        else:
+            msg = resp.get('MESSAGE', '')
+            raise I2PCommandFail(result, msg, "STREAM CONNECT")
+    
+    #@-node:samStreamConnect
+    #@+node:samStreamAccept
+    def samStreamAccept(self):
+        """
+        Waits for an incoming connection, returning a wrapped conn obj
+        """
+        log(4, "waiting for connection")
+        conn = self.qNewStreams.get()
+        log(4, "got connection")
+        return conn
+    #@-node:samStreamAccept
+    #@+node:samStreamSend
+    def samStreamSend(self, conn, data):
+        """
+        DO NOT CALL THIS DIRECTLY
+        
+        Invoked by an I2PSAMStream object to transfer data
+        Use the object's .send() method instead.
+        
+        conn is the I2PSAMStream
+    
+        STREAM SEND 
+        ID=$id
+        SIZE=$numBytes\n[$numBytes of data]
+        """
+        # dispatch
+        self.samSend("STREAM", "SEND", data, ID=conn.id)
+    
+        # useless, but mimics socket paradigm
+        return len(data)
+    
+    #@-node:samStreamSend
+    #@+node:samStreamClose
+    def samStreamClose(self, conn):
+        """
+        DO NOT CALL DIRECTLY
+        
+        Invoked by I2PSAMStream to close stream
+        Use the object's .send() method instead.
+        
+        STREAM CLOSE
+        ID=$id
+        """
+        self.samSend("STREAM", "CLOSE", ID=conn.id)
+        del self.streams[conn.id]
+    
+    #@-node:samStreamClose
     #@+node:samNamingLookup
     def samNamingLookup(self, host):
         """
@@ -472,7 +591,7 @@ class I2PSamClient:
             try:
                 name, val = arg.split("=", 1)
             except:
-                logException(3, "failed to process %s" % repr(arg))
+                logException(3, "failed to process %s in %s" % (repr(arg), repr(flds)))
                 raise
             dargs[name] = val
     
@@ -488,6 +607,8 @@ class I2PSamClient:
     
     
     
+    
+    
     #@-node:samParse
     #@+node:samSend
     def samSend(self, topic, subtopic, data=None, **kw):
@@ -586,7 +707,90 @@ class I2PSamClient:
     def on_STREAM(self, topic, subtopic, args):
         """
         Handles STREAM messages from server
+    
+        STREAM STATUS
+        RESULT=$result
+        ID=$id
+        [MESSAGE=...]
+    
+        STREAM CONNECTED
+        DESTINATION=$base64key
+        ID=$id
+    
+        STREAM RECEIVED
+        ID=$id
+        SIZE=$numBytes\n[$numBytes of data]
+    
+        STREAM CLOSED
+        RESULT=$result
+        ID=$id
+        [MESSAGE=...]
         """
+        log(4, "got %s %s %s" % (topic, subtopic, args))
+    
+        # which stream?
+        id = int(args['ID'])
+    
+        # result of prior connection attempt
+        if subtopic == 'STATUS':
+            # stick it on the queue that the caller is waiting on and let the
+            # caller interpret the result
+            self.streamConnectReplies[id].put(args)
+            return 
+    
+        # notice of incoming connection
+        if subtopic == 'CONNECTED':
+    
+            # grab details
+            dest = args['DESTINATION']
+    
+            # wrap it in a stream obj
+            conn = I2PSAMStream(self, id, dest)
+            self.streams[id] = conn
+    
+            # and put it there for anyone calling samStreamAccept()
+            self.qNewStreams.put(conn)
+    
+            # done
+            return
+    
+        # notice of received data
+        elif subtopic == 'RECEIVED':
+            # grab details
+            data = args['DATA']
+    
+            # lookup the connection
+            conn = self.streams.get(id, None)
+            if not conn:
+                # conn not known, just ditch
+                log(2, "got data, but don't recall any conn with id %s" % id)
+                return
+    
+            # and post the received data
+            conn._notifyIncomingData(data)
+    
+            log(4, "wrote data to conn's inbound queue")
+            
+            # done
+            return
+    
+        elif subtopic == 'CLOSED':
+            # lookup the connection
+            conn = self.streams.get(id, None)
+            if not conn:
+                # conn not known, just ditch
+                return
+    
+            # mark conn as closed and forget it
+            conn._notifyIncomingData("") # special signal to close
+            conn.isOpen = 0
+            del self.streams[id]
+    
+            # done
+            return
+    
+    
+    
     #@-node:on_STREAM
     #@+node:on_DATAGRAM
     def on_DATAGRAM(self, topic, subtopic, args):
@@ -627,6 +831,17 @@ class I2PSamClient:
         res = pubkey, privkey
         self.qNewDests.put(res)
     #@-node:on_DEST
+    #@+node:samAllocId
+    def samAllocId(self):
+        """
+        Allocates a new unique id as required by SAM protocol
+        """
+        self.samNextIdLock.acquire()
+        id = self.samNextId
+        self.samNextId += 1
+        self.samNextIdLock.release()
+        return id
+    #@-node:samAllocId
     #@+node:_recvline
     def _recvline(self):
         """
@@ -680,6 +895,117 @@ class I2PSamClient:
     #@-node:_sendline
     #@-others
 #@-node:class I2PSamClient
+#@+node:class I2PSAMStream
+class I2PSAMStream:
+    """
+    Wrapper for a stream object
+    """
+    #@    @+others
+    #@+node:__init__
+    def __init__(self, client, id, dest):
+        """
+        """
+        self.client = client
+        self.id = id
+        self.dest = dest
+    
+        self.qIncomingData = Queue.Queue()
+    
+        self.inbuf = ''
+        self.isOpen = 1
+    #@-node:__init__
+    #@+node:send
+    def send(self, data):
+        """
+        Sends data to a stream connection
+        """
+        # barf if stream not open
+        if not self.isOpen:
+            raise I2PStreamClosed
+    
+        # can send
+        return self.client.samStreamSend(self, data)
+    #@-node:send
+    #@+node:recv
+    def recv(self, size):
+        """
+        Retrieves n bytes from peer
+        """
+        chunks = []
+    
+        while self.isOpen and size > 0:
+            # try internal buffer first
+            if self.inbuf:
+                chunk = self.inbuf[:size]
+                chunklen = len(chunk)
+                self.inbuf = self.inbuf[chunklen:]
+                chunks.append(chunk)
+                size -= chunklen
+            else:
+                # replenish input buffer
+                log(4, "I2PSAMStream.recv: replenishing input buffer")
+                buf = self.qIncomingData.get()
+                if buf == '':
+                    # connection closed by peer
+                    self.isOpen = 0
+                    break
+                else:
+                    # got more data
+                    log(4, "I2PSAMStream: queue returned %s" % repr(buf))
+                    self.inbuf += buf
+    
+        # return whatever we've got, hopefully all
+        return "".join(chunks)
+    
+    
+    #@-node:recv
+    #@+node:readline
+    def readline(self):
+        """
+        Read a line of text from stream, return the line without trailing newline
+        
+        This method really shouldn't exist in a class that's trying to look a bit
+        like a socket object, but what the hell!
+        """
+        chars = []
+        while 1:
+            char = self.recv(1)
+            if char in ['', '\n']:
+                break
+            chars.append(char)
+        return "".join(chars)
+    #@-node:readline
+    #@+node:close
+    def close(self):
+        """
+        close this stream connection
+        """
+        log(4, "closing stream")
+        self.client.samStreamClose(self)
+        log(4, "stream closed")
+        self.isOpen = 0
+    
+        # and just to make sure...
+        self.qIncomingData.put("") # busts out of recv() loops
+    
+    #@-node:close
+    #@+node:__del__
+    def __del__(self):
+        """
+        Dropping last ref to this object closes stream
+        """
+        self.close()
+    #@-node:__del__
+    #@+node:_notifyIncomingData
+    def _notifyIncomingData(self, data):
+        """
+        Called by client receiver to notify incoming data
+        """
+        log(4, "got %s" % repr(data))
+        self.qIncomingData.put(data)
+    #@-node:_notifyIncomingData
+    #@-others
+#@-node:class I2PSAMStream
 #@+node:class I2PRemoteSession
 class I2PRemoteSession:
     """
@@ -701,7 +1027,8 @@ class I2PRemoteSession:
     #@-node:__init__
     #@+node:send
     def send(self, peerdest, msg):
-    
+        """
+        """
         return self.client.send(self.dest, peerdest, msg)
     #@-node:send
     #@+node:recv
@@ -885,35 +1212,59 @@ def demoSTREAM():
     print "Starting SAM STREAM demo..."
     print
 
-    print "Instantiating 2 more client connections..."
+    print "Instantiating client c6..."
-    c5 = I2PSamClient()
     c6 = I2PSamClient()
 
-    print "Creating more dests via SAM"
+    print "Creating dest for c6"
-    pub5, priv5 = c5.samDestGenerate()
     pub6, priv6 = c6.samDestGenerate()
 
-    print "Creating SAM STREAM SESSION on connection c3..."
+    print "Creating SAM STREAM SESSION on connection c6..."
-    res = c5.samSessionCreate("STREAM", priv5)
+    res = c6.samSessionCreate("STREAM", priv6, direction="RECEIVE")
+    if res != 'OK':
+        print "Failed to create STREAM session on connection c6: %s" % repr(res)
+        return
+    print "STREAM Session on connection c6 created successfully"
+
+    print "Launching acceptor thread..."
+    thread.start_new_thread(demoSTREAM_thread, (c6,))
+
+    #print "sleep a while and give the server a chance..."
+    #time.sleep(10)
+
+    print "----------------------------------------"
+
+    print "Instantiating client c5..."
+    c5 = I2PSamClient()
+
+    print "Creating dest for c5"
+    pub5, priv5 = c5.samDestGenerate()
+
+    print "Creating SAM STREAM SESSION on connection c5..."
+    res = c5.samSessionCreate("STREAM", priv5, direction="CREATE")
     if res != 'OK':
         print "Failed to create STREAM session on connection c5: %s" % repr(res)
         return
     print "STREAM Session on connection c5 created successfully"
 
-    print "Creating SAM STREAM SESSION on connection c6..."
+    print "----------------------------------------"
-    res = c6.samSessionCreate("STREAM", priv6)
+
-    if res != 'OK':
+    print "Making connection from c5 to c6..."
-        print "Failed to create STREAM session on connection c4: %s" % repr(res)
+
+    #set_trace()
+
+    try:
+        conn_c5 = c5.samStreamConnect(pub6)
+    except:
+        print "Stream Connection failed"
         return
-    print "STREAM Session on connection c4 created successfully"
+    print "Stream connect succeeded"
+    
+    print "Receiving from c5..."
+    buf = conn_c5.readline()
+    print "Got %s" % repr(buf)
 
-    msg = "Hi there, this is a datagram!"
+    #print "Try to accept connection on c6..."
-    print "sending from c5 to c6: %s" % repr(msg)
+    #conn_c6 = c6.sam
-    c5.samStreamSend(pub6, msg)
-
-    print "now try to receive from c6 (will block)..."
-    msg1 = c6.samStreamReceive()
-    print "Connection c6 got %s from %s..." % (repr(msg1), repr(remdest))
 
     print
     print "--------------------------------------"
@@ -921,7 +1272,31 @@ def demoSTREAM():
     print "--------------------------------------"
     print
 
+
+
+
+
 #@-node:demoSTREAM
+#@+node:demoSTREAM_thread
+def demoSTREAM_thread(sess):
+    
+    while 1:
+        sock = sess.samStreamAccept()
+        log(4, "got incoming connection")
+
+        print "**ACCEPTOR SLEEPING 10 secs BEFORE SENDING"
+        
+        time.sleep(10)
+
+        sock.send("Hi there, what do you want?\n")
+
+        print "**ACCEPTOR SLEEPING 5 MINS BEFORE CLOSING"
+        time.sleep(300)
+        print "**ACCEPTOR CLOSING STREAM"
+
+        sock.close()
+
+#@-node:demoSTREAM_thread
 #@+node:demo
 def demo():
     """
@@ -936,10 +1311,10 @@ def demo():
     print "-----------------------------------------"
     print
 
-    demoNAMING()
+    #demoNAMING()
-    demoRAW()
+    #demoRAW()
-    demoDATAGRAM()
+    #demoDATAGRAM()
-    #demoSTREAM()
+    demoSTREAM()
 
     print
     print "-----------------------------------------"