Various updates and fixes:

Separate endpoint initialization from opening.
Allow remote to name the link.
Fix examples to index using container+link name.
Clean up connection processing state.

examples/python/rpc-client.py

@@ -37,16 +37,34 @@ to a server, and waits for a response.  The method call is a map of the form:
 
 class MyConnection(fusion.ConnectionEventHandler):
 
-    def __init__(self, name, socket, container, properties):
+    def __init__(self, name, container, properties):
         self.name = name
-        self.socket = socket
+        self.container = container
-        self.connection = container.create_connection(name, self,
+        self.properties = properties
-                                                      properties)
+        self.socket = None
-        self.connection.user_context = self
+        self.caller = None
+        self.connection = None
 
+    def reset(self):
+        if self.caller:
+            self.caller.reset()
+        if self.connection:
+            self.connection.user_context = None
+            self.connection.destroy()
+            self.connection = None
+        if self.socket:
+            self.socket.close()
+            self.socket = None
+
+    def connect(self, socket):
+        self.reset()
+        self.socket = socket
+        self.connection = self.container.create_connection(self.name,
+                                                           self,
+                                                           self.properties)
+        self.connection.user_context = self
         self.connection.sasl.mechanisms("ANONYMOUS")
         self.connection.sasl.client()
-
         self.connection.open()
 
     def process(self):
@@ -84,11 +102,20 @@ class MyConnection(fusion.ConnectionEventHandler):
         return self.connection.closed
 
     def destroy(self, error=None):
-        self.connection.user_context = None
+        self.reset()
-        self.connection.destroy()
+        self.caller.destroy()
-        self.connection = None
+        self.caller = None
-        self.socket.close()
+        self.container = None
-        self.socket = None
+
+    def create_caller(self, method_map, source_addr, target_addr,
+                      receiver_properties, sender_properties):
+        """ Caller factory
+        """
+        if self.caller:
+            self.caller.destroy()
+        self.caller = MyCaller(method_map, self, source_addr, target_addr,
+                               receiver_properties, sender_properties)
+        return self.caller
 
     # Connection callbacks:
 
@@ -96,7 +123,10 @@ class MyConnection(fusion.ConnectionEventHandler):
         """connection handshake completed"""
         LOG.debug("Connection active callback")
 
-    def connection_closed(self, connection, reason):
+    def connection_remote_closed(self, connection, reason):
+        LOG.debug("Connection remote closed callback")
+
+    def connection_closed(self, connection):
         LOG.debug("Connection closed callback")
 
     def sender_requested(self, connection, link_handle,
@@ -123,20 +153,48 @@ class MyCaller(fusion.SenderEventHandler,
     def __init__(self, method_map, my_connection,
                  my_source, my_target,
                  receiver_properties, sender_properties):
-        conn = my_connection.connection
+        #self._name = uuid.uuid4().hex
-        self._sender = conn.create_sender(my_source, target_address=None,
+        self._my_connection = my_connection
-                                          eventHandler=self, name=my_source,
+        self._source_addr = my_source
-                                          properties=sender_properties)
+        self._target_addr = my_target
-        self._receiver = conn.create_receiver(my_target, source_address=None,
+        self._receiver_properties = receiver_properties
-                                              eventHandler=self, name=my_target,
+        self._sender_properties = sender_properties
-                                              properties=receiver_properties)
         self._method = method_map
+        self._sender = None
+        self._receiver = None
+        self.reset()
+
+    def reset(self):
+        LOG.debug("Resetting my-caller")
+        # @todo: for now, use new name as engine isn't cleaning up link state properly...
+        self._name = uuid.uuid4().hex
         self._reply_to = None
         self._to = None
-
         self._send_completed = False
         self._response_received = False
 
+        if self._sender:
+            self._sender.destroy()
+            self._sender = None
+
+        if self._receiver:
+            self._receiver.destroy()
+            self._receiver = None
+
+    def connect(self):
+        self.reset()
+        LOG.debug("Connecting my-caller")
+        conn = self._my_connection.connection
+        self._sender = conn.create_sender(self._source_addr, target_address=None,
+                                          eventHandler=self,
+                                          #name=self._source_addr,
+                                          name=self._name,
+                                          properties=self._sender_properties)
+        self._receiver = conn.create_receiver(self._target_addr, source_address=None,
+                                              eventHandler=self,
+                                              #name=self._target_addr,
+                                              name=self._name,
+                                              properties=self._receiver_properties)
         self._sender.open()
         self._receiver.add_capacity(1)
         self._receiver.open()
@@ -145,14 +203,17 @@ class MyCaller(fusion.SenderEventHandler,
         return self._send_completed and self._response_received
 
     def close(self):
+        LOG.debug("Closing my-caller")
         self._sender.close(None)
         self._receiver.close(None)
 
+    def closed(self):
+        return self._sender.closed and self._receiver.closed
+
     def destroy(self):
-        if self._sender:
+        LOG.debug("Destroying my-caller")
-            self._sender.destroy()
+        self.reset()
-        if self._receiver:
+        self._my_connection = None
-            self._receiver.destroy()
 
     def _send_request(self):
         """Send a message containing the RPC method call
@@ -179,7 +240,12 @@ class MyCaller(fusion.SenderEventHandler,
         if self._reply_to:
             self._send_request()
 
-    def sender_closed(self, sender_link, error):
+    def sender_remote_closed(self, sender_link, error):
+        LOG.debug("Sender remote closed callback")
+        assert sender_link is self._sender
+        self._sender.close()
+
+    def sender_closed(self, sender_link):
         LOG.debug("Sender closed callback")
         assert sender_link is self._sender
         self._send_completed = True
@@ -200,7 +266,12 @@ class MyCaller(fusion.SenderEventHandler,
         if self._to:
             self._send_request()
 
-    def receiver_closed(self, receiver_link, error):
+    def receiver_remote_closed(self, receiver_link, error):
+        LOG.debug("receiver remote closed callback")
+        assert receiver_link is self._receiver
+        self._receiver.close()
+
+    def receiver_closed(self, receiver_link):
         LOG.debug("Receiver closed callback")
         assert receiver_link is self._receiver
         self._response_received = True
@@ -222,6 +293,9 @@ def main(argv=None):
                       help="The address of the server [amqp://0.0.0.0:5672]")
     parser.add_option("-t", "--timeout", dest="timeout", type="int",
                       help="timeout used when waiting for reply, in seconds")
+    parser.add_option("--repeat", dest="repeat", type="int",
+                      default=1,
+                      help="Repeat the RPC call REPEAT times (0 == forever)")
     parser.add_option("--trace", dest="trace", action="store_true",
                       help="enable protocol tracing")
     parser.add_option("--debug", dest="debug", action="store_true",
@@ -263,26 +337,42 @@ def main(argv=None):
     conn_properties = {}
     if opts.trace:
         conn_properties["trace"] = True
-    my_connection = MyConnection( "to-server", my_socket,
+
-                                  container, conn_properties)
+    my_connection = MyConnection( "to-server", container, conn_properties)
+
 
     # Create the RPC caller
     method = {'method': method_info[0],
               'args': dict([(method_info[i], method_info[i+1])
                             for i in range(1, len(method_info), 2)])}
-    my_caller = MyCaller( method,
+    my_caller = my_connection.create_caller( method,
-                          my_connection,
+                                             "my-source-address",
-                          "my-source-address",
+                                             "my-target-address",
-                          "my-target-address",
+                                             receiver_properties={},
-                          receiver_properties={"capacity": 1},
+                                             sender_properties={})
-                          sender_properties={})
+    my_connection.connect(my_socket)
 
-    while not my_caller.done():
+    repeat = 0
-        my_connection.process()
+    while opts.repeat == 0 or repeat < opts.repeat:
 
-    LOG.debug("RPC completed, closing connections")
+        LOG.debug("Requesting RPC...")
 
-    my_caller.close()
+        my_caller.connect()
+        while not my_caller.done():
+            my_connection.process()
+
+        LOG.debug("RPC completed!  Closing caller...")
+
+        my_caller.close()
+
+        while not my_caller.closed():
+            my_connection.process()
+
+        LOG.debug("Caller closed cleanly!")
+
+        repeat += 1
+
+    print("Closing connections")
     my_connection.close()
     while not my_connection.closed:
         my_connection.process()

examples/python/rpc-server.py

@@ -38,9 +38,16 @@ map sent in the request.
 """
 
 
-# Maps of outgoing and incoming links
+# Maps of outgoing and incoming links.  These are indexed by
-sender_links = {}  # indexed by Source address
+# (remote-container-name, link-name)
-receiver_links = {} # indexed by Target address
+sender_links = {}
+receiver_links = {}
+
+# Map reply-to address to the proper sending link (indexed by address)
+reply_senders = {}
+
+# database of all active SocketConnections
+socket_connections = {} # indexed by name
 
 
 class SocketConnection(fusion.ConnectionEventHandler):
@@ -55,6 +62,16 @@ class SocketConnection(fusion.ConnectionEventHandler):
         self.connection.sasl.mechanisms("ANONYMOUS")
         self.connection.sasl.server()
         self.connection.open()
+        self.done = False
+
+    def destroy(self):
+        self.done = True
+        if self.connection:
+            self.connection.destroy()
+            self.connection = None
+        if self.socket:
+            self.socket.close()
+            self.socket = None
 
     def fileno(self):
         """Allows use of a SocketConnection in a select() call.
@@ -80,43 +97,58 @@ class SocketConnection(fusion.ConnectionEventHandler):
     def connection_active(self, connection):
         LOG.debug("Connection active callback")
 
-    def connection_closed(self, connection, reason):
+    def connection_remote_closed(self, connection, reason):
-        LOG.debug("Connection closed callback")
+        LOG.debug("Connection remote closed callback")
+        assert self.connection is connection
+        self.connection.close()
+
+    def connection_closed(self, connection):
+        LOG.debug("connection closed.")
+        # main loop will destroy
+        self.done = True
 
     def sender_requested(self, connection, link_handle,
-                         requested_source, properties):
+                         name, requested_source, properties):
         LOG.debug("sender requested callback")
         global sender_links
+        global reply_senders
+
+        # reject if name conflict
+        remote_container = connection.remote_container
+        ident = (remote_container, name)
+        if ident in sender_links:
+            connection.reject_sender(link_handle, "link name in use")
+            return
 
-        name = uuid.uuid4().hex
         # allow for requested_source address if it doesn't conflict with an
-        # existing address
+        # existing address, otherwise override
-        if not requested_source or requested_source in sender_links:
+        if not requested_source or requested_source in reply_senders:
-            requested_source = "/%s/%s" % (connection.container.name,
+            requested_source = uuid.uuid4().hex
-                                           name)
+            assert requested_source not in reply_senders
-            assert requested_source not in sender_links
 
-        sender = MySenderLink(connection, link_handle, requested_source,
+        sender = MySenderLink(ident, connection, link_handle, requested_source)
-                              name, {})
+        sender_links[ident] = sender
-        sender_links[requested_source] = sender
+        reply_senders[requested_source] = sender
         print("New Sender link created, source=%s" % requested_source)
 
     def receiver_requested(self, connection, link_handle,
-                           requested_target, properties):
+                           name, requested_target, properties):
         LOG.debug("receiver requested callback")
-        # allow for requested_source address if it doesn't conflict with an
-        # existing address
         global receiver_links
 
-        name = uuid.uuid4().hex
+        # reject if name conflict
-        if not requested_target or requested_target in receiver_links:
+        remote_container = connection.remote_container
-            requested_target = "/%s/%s" % (connection.container.name,
+        ident = (remote_container, name)
-                                           name)
+        if ident in receiver_links:
-            assert requested_target not in receiver_links
+            connection.reject_sender(link_handle, "link name in use")
+            return
 
-        receiver = MyReceiverLink(connection, link_handle,
+        # I don't use the target address, but supply one if necessary
-                                  requested_target, name)
+        if not requested_target:
-        receiver_links[requested_target] = receiver
+            requested_target = uuid.uuid4().hex
+
+        receiver = MyReceiverLink(ident, connection, link_handle, requested_target)
+        receiver_links[ident] = receiver
         print("New Receiver link created, target=%s" % requested_target)
 
     # SASL callbacks:
@@ -132,13 +164,14 @@ class SocketConnection(fusion.ConnectionEventHandler):
 class MySenderLink(fusion.SenderEventHandler):
     """
     """
-    def __init__(self, connection, link_handle, source_address,
+    def __init__(self, ident, connection, link_handle,
-                 name, properties):
+                 source_address, properties={}):
 
+        self._ident = ident
+        self._source_address = source_address
         self.sender_link = connection.accept_sender(link_handle,
                                                     source_address,
                                                     self,
-                                                    name,
                                                     properties)
         self.sender_link.open()
 
@@ -147,8 +180,21 @@ class MySenderLink(fusion.SenderEventHandler):
     def sender_active(self, sender_link):
         LOG.debug("sender active callback")
 
-    def sender_closed(self, sender_link, error):
+    def sender_remote_closed(self, sender_link, error):
+        LOG.debug("sender remote closed callback")
+        self.sender_link.close()
+
+    def sender_closed(self, sender_link):
         LOG.debug("sender closed callback")
+        global sender_links
+        global reply_senders
+
+        if self._ident in sender_links:
+            del sender_links[self._ident]
+        if self._source_address in reply_senders:
+            del reply_senders[self._source_address]
+        self.sender_link.destroy()
+        self.sender_link = None
 
     # 'message sent' callback:
 
@@ -159,12 +205,13 @@ class MySenderLink(fusion.SenderEventHandler):
 class MyReceiverLink(fusion.ReceiverEventHandler):
     """
     """
-    def __init__(self, connection, link_handle, target_address,
+    def __init__(self, ident, connection, link_handle, target_address,
-                 name, properties={}):
+                 properties={}):
+        self._ident = ident
+        self._target_address = target_address
         self._link = connection.accept_receiver(link_handle,
                                                 target_address,
                                                 self,
-                                                name,
                                                 properties)
         self._link.open()
 
@@ -173,22 +220,32 @@ class MyReceiverLink(fusion.ReceiverEventHandler):
         LOG.debug("receiver active callback")
         self._link.add_capacity(5)
 
-    def receiver_closed(self, receiver_link, error):
+    def receiver_remote_closed(self, receiver_link, error):
+        LOG.debug("receiver remote closed callback")
+        self._link.close()
+
+    def receiver_closed(self, receiver_link):
         LOG.debug("receiver closed callback")
+        global receiver_links
+
+        if self._ident in receiver_links:
+            del receiver_links[self._ident]
+        self._link.destroy()
+        self._link = None
 
     def message_received(self, receiver_link, message, handle):
         LOG.debug("message received callback")
 
-        global sender_links
+        global reply_senders
 
         # extract to reply-to, correlation id
         reply_to = message.reply_to
-        if not reply_to or reply_to not in sender_links:
+        if not reply_to or reply_to not in reply_senders:
             LOG.error("sender for reply-to not found, reply-to=%s",
                       str(reply_to))
             self._link.message_rejected(handle, "Bad reply-to address")
         else:
-            my_sender = sender_links[reply_to]
+            my_sender = reply_senders[reply_to]
             correlation_id = message.correlation_id
             method_map = message.body
             if (not isinstance(method_map, dict) or
@@ -260,7 +317,7 @@ def main(argv=None):
     # create an AMQP container that will 'provide' the RPC service
     #
     container = fusion.Container("example RPC service")
-    socket_connections = {} # indexed by name (uuid)
+    global socket_connections
 
     while True:
 
@@ -292,6 +349,7 @@ def main(argv=None):
         readable,writable,ignore = select.select(readfd,writefd,[],timeout)
         LOG.debug("select() returned")
 
+        worked = []
         for r in readable:
             if r is my_socket:
                 # new inbound connection request
@@ -305,19 +363,38 @@ def main(argv=None):
                                                             client_socket,
                                                             container,
                                                             conn_properties)
+                LOG.debug("new connection created name=%s" % name)
+
             else:
                 assert isinstance(r, SocketConnection)
                 rc = r.process_input()
+                worked.append(r)
 
         for t in timers:
             now = time.time()
             if t.next_tick > now:
                 break
             t.process(now)
+            sc = t.user_context
+            assert isinstance(sc, SocketConnection)
+            worked.append(sc)
 
         for w in writable:
             assert isinstance(w, SocketConnection)
             rc = w.send_output()
+            worked.append(w)
+
+        # nuke any completed connections:
+        closed = False
+        while worked:
+            sc = worked.pop()
+            if sc.done:
+                if sc.name in socket_connections:
+                    del socket_connections[sc.name]
+                sc.destroy()
+                closed = True
+        if closed:
+            LOG.debug("%d active connections present" % len(socket_connections))
 
     return 0
 

python/connection.py

@@ -35,17 +35,22 @@ class ConnectionEventHandler(object):
         """connection handshake completed"""
         LOG.debug("connection_active (ignored)")
 
-    def connection_closed(self, connection, reason):
+    def connection_remote_closed(self, connection, error=None):
+        LOG.debug("connection_remote_closed (ignored)")
+
+    def connection_closed(self, connection):
         LOG.debug("connection_closed (ignored)")
 
     def sender_requested(self, connection, link_handle,
-                         requested_source, properties={}):
+                         name, requested_source,
+                         properties={}):
         # call accept_sender to accept new link,
         # reject_sender to reject it.
         LOG.debug("sender_requested (ignored)")
 
     def receiver_requested(self, connection, link_handle,
-                           requested_target, properties={}):
+                           name, requested_target,
+                           properties={}):
         # call accept_sender to accept new link,
         # reject_sender to reject it.
         LOG.debug("receiver_requested (ignored)")
@@ -81,10 +86,10 @@ class Connection(object):
         if properties.get("trace"):
             self._pn_transport.trace(proton.Transport.TRACE_FRM)
 
-        self._sender_links = {}
+        # indexed by link-name
-        self._receiver_links = {}
+        self._sender_links = {}    # SenderLink or pn_link if pending
-        self._pending_links = {}
+        self._receiver_links = {}  # ReceiverLink or pn_link if pending
-        self._pending_link_id = 0
+
         self._read_done = False
         self._write_done = False
         self._next_tick = 0
@@ -101,18 +106,25 @@ class Connection(object):
 
     @property
     # @todo - hopefully remove
-    def transport(self):
+    def pn_transport(self):
         return self._pn_transport
 
     @property
     # @todo - hopefully remove
-    def connection(self):
+    def pn_connection(self):
         return self._pn_connection
 
     @property
     def name(self):
         return self._name
 
+    @property
+    def remote_container(self):
+        """Return the name of the remote container. Should be present once the
+        connection is active.
+        """
+        return self._pn_connection.remote_container
+
     @property
     # @todo - think about server side use of sasl!
     def sasl(self):
@@ -131,9 +143,73 @@ class Connection(object):
 Associate an arbitrary user object with this Connection.
 """)
 
-    _NEED_INIT = proton.Endpoint.LOCAL_UNINIT
+    def open(self):
-    _NEED_CLOSE = (proton.Endpoint.LOCAL_ACTIVE|proton.Endpoint.REMOTE_CLOSED)
+        """
+        """
+        self._pn_connection.open()
+        self._pn_session = self._pn_connection.session()
+        self._pn_session.open()
+
+    def close(self, error=None):
+        """
+        """
+        for l in self._sender_links.itervalues():
+            l.close(error)
+        for l in self._receiver_links.itervalues():
+            l.close(error)
+        self._pn_session.close()
+        self._pn_connection.close()
+
+    @property
+    def closed(self):
+        #return self._write_done and self._read_done
+        state = self._pn_connection.state
+        return state == (proton.Endpoint.LOCAL_CLOSED
+                         | proton.Endpoint.REMOTE_CLOSED)
+
+    def destroy(self):
+        """
+        """
+        self._sender_links.clear()
+        self._receiver_links.clear()
+        self._container._remove_connection(self._name)
+        self._container = None
+        self._pn_connection = None
+        self._pn_transport = None
+        self._user_context = None
+
+    def _link_requested(self, pn_link):
+        if pn_link.is_sender and pn_link.name not in self._sender_links:
+            LOG.debug("Remotely initiated Sender needs init")
+            self._sender_links[pn_link.name] = pn_link
+            pn_link.context = None # @todo: update proton.py
+            req_source = ""
+            if pn_link.remote_source.dynamic:
+                req_source = None
+            elif pn_link.remote_source.address:
+                req_source = pn_link.remote_source.address
+            self._handler.sender_requested(self, pn_link.name,
+                                           pn_link.name, req_source,
+                                           {"target-address":
+                                            pn_link.remote_target.address})
+        elif pn_link.is_receiver and pn_link.name not in self._receiver_links:
+            LOG.debug("Remotely initiated Receiver needs init")
+            self._receiver_links[pn_link.name] = pn_link
+            pn_link.context = None # @todo: update proton.py
+            req_target = ""
+            if pn_link.remote_target.dynamic:
+                req_target = None
+            elif pn_link.remote_target.address:
+                req_target = pn_link.remote_target.address
+            self._handler.receiver_requested(self, pn_link.name,
+                                             pn_link.name, req_target,
+                                             {"source-address":
+                                              pn_link.remote_source.address})
+
+    _REMOTE_REQ = (proton.Endpoint.LOCAL_UNINIT|proton.Endpoint.REMOTE_ACTIVE)
+    _REMOTE_CLOSE = (proton.Endpoint.LOCAL_ACTIVE|proton.Endpoint.REMOTE_CLOSED)
     _ACTIVE = (proton.Endpoint.LOCAL_ACTIVE|proton.Endpoint.REMOTE_ACTIVE)
+    _CLOSED = (proton.Endpoint.LOCAL_CLOSED|proton.Endpoint.REMOTE_CLOSED)
 
     def process(self, now):
         """
@@ -152,103 +228,112 @@ Associate an arbitrary user object with this Connection.
             self._handler.sasl_done(self, self._pn_sasl.outcome)
             self._pn_sasl = None
 
-        if self._pn_connection.state & self._NEED_INIT:
+        # do endpoint up handling:
-            assert False, "Connection always opened() on create"
 
-        if not self._active:
+        if self._pn_connection.state == self._ACTIVE:
-            if self._pn_connection.state == self._ACTIVE:
+            if not self._active:
                 self._active = True
                 self._handler.connection_active(self)
 
-        ssn = self._pn_connection.session_head(self._NEED_INIT)
+        pn_session = self._pn_connection.session_head(proton.Endpoint.LOCAL_UNINIT)
-        while ssn:
+        while pn_session:
             LOG.debug("Opening remotely initiated session")
-            ssn.open()
+            pn_session.open()
-            ssn = ssn.next(self._NEED_INIT)
+            pn_session = pn_session.next(proton.Endpoint.LOCAL_UNINIT)
 
-        link = self._pn_connection.link_head(self._NEED_INIT)
+        pn_link = self._pn_connection.link_head(self._REMOTE_REQ)
-        while link:
+        while pn_link:
-            LOG.debug("Remotely initiated Link needs init")
+            next_link = pn_link.next(proton.Endpoint.LOCAL_UNINIT)
-            index = self._pending_link_id
-            self._pending_link_id += 1
-            assert index not in self._pending_links
-            self._pending_links[index] = link
-            if link.is_sender:
-                req_source = ""
-                if link.remote_source.dynamic:
-                    req_source = None
-                elif link.remote_source.address:
-                    req_source = link.remote_source.address
-                self._handler.sender_requested(self, index, req_source,
-                                               {"target-address":
-                                                link.remote_target.address})
-            else:
-                req_target = ""
-                if link.remote_target.dynamic:
-                    req_target = None
-                elif link.remote_target.address:
-                    req_target = link.remote_target.address
-                self._handler.receiver_requested(self, index, req_target,
-                                                 {"source-address":
-                                                  link.remote_source.address})
-            link = link.next(self._NEED_INIT)
 
-        # @todo: won't scale
+            if pn_link.state == self._REMOTE_REQ:
-        link = self._pn_connection.link_head(self._ACTIVE)
+                self._link_requested(pn_link)
-        while link:
+
-            if link.context and not link.context._active:
+            pn_link = next_link
-                if link.is_sender:
+
-                    sender_link = link.context
+        # @todo: won't scale?
+        pn_link = self._pn_connection.link_head(self._ACTIVE)
+        while pn_link:
+            next_link = pn_link.next(self._ACTIVE)
+
+            if pn_link.context and not pn_link.context._active:
+                LOG.debug("Link is up")
+                pn_link.context._active = True
+                if pn_link.is_sender:
+                    sender_link = pn_link.context
+                    assert isinstance(sender_link, SenderLink)
                     sender_link._handler.sender_active(sender_link)
                 else:
-                    receiver_link = link.context
+                    receiver_link = pn_link.context
+                    assert isinstance(receiver_link, ReceiverLink)
                     receiver_link._handler.receiver_active(receiver_link)
-                link.context._active = True
+            pn_link = next_link
-            link = link.next(self._ACTIVE)
 
         # process the work queue
 
-        delivery = self._pn_connection.work_head
+        pn_delivery = self._pn_connection.work_head
-        while delivery:
+        while pn_delivery:
             LOG.debug("Delivery updated!")
-            if delivery.link.context:
+            next_delivery = pn_delivery.work_next
-                if delivery.link.is_sender:
+            if pn_delivery.link.context:
-                    sender_link = delivery.link.context
+                if pn_delivery.link.is_sender:
-                    sender_link._delivery_updated(delivery)
+                    sender_link = pn_delivery.link.context
+                    sender_link._delivery_updated(pn_delivery)
                 else:
-                    receiver_link = delivery.link.context
+                    receiver_link = pn_delivery.link.context
-                    receiver_link._delivery_updated(delivery)
+                    receiver_link._delivery_updated(pn_delivery)
-            delivery = delivery.work_next
+            pn_delivery = next_delivery
 
-        # close all endpoints closed by remotes
+        # do endpoint down handling:
 
-        link = self._pn_connection.link_head(self._NEED_CLOSE)
+        pn_link = self._pn_connection.link_head(self._REMOTE_CLOSE)
-        while link:
+        while pn_link:
             LOG.debug("Link closed remotely")
-            link.close()
+            next_link = pn_link.next(self._REMOTE_CLOSE)
             # @todo: error reporting
-            if link.context:
+            if pn_link.context:
-                if link.is_sender:
+                if pn_link.is_sender:
-                    sender_link = link.context
+                    sender_link = pn_link.context
-                    sender_link._handler.sender_closed(sender_link, None)
+                    sender_link._handler.sender_remote_closed(sender_link, None)
                 else:
-                    receiver_link = link.context
+                    receiver_link = pn_link.context
-                    receiver_link._handler.receiver_closed(receiver_link,
+                    receiver_link._handler.receiver_remote_closed(receiver_link,
-                                                           None)
+                                                                  None)
-            link = link.next(self._NEED_CLOSE)
+            pn_link = next_link
 
-        ssn = self._pn_connection.session_head(self._NEED_CLOSE)
+        pn_link = self._pn_connection.link_head(self._CLOSED)
-        while ssn:
+        while pn_link:
+            next_link = pn_link.next(self._CLOSED)
+            if pn_link.context and pn_link.context._active:
+                LOG.debug("Link close completed")
+                pn_link.context._active = False
+                if pn_link.is_sender:
+                    sender_link = pn_link.context
+                    sender_link._handler.sender_closed(sender_link)
+                else:
+                    receiver_link = pn_link.context
+                    receiver_link._handler.receiver_closed(receiver_link)
+            pn_link = next_link
+
+        pn_session = self._pn_connection.session_head(self._REMOTE_CLOSE)
+        while pn_session:
             LOG.debug("Session closed remotely")
-            ssn.close()
+            pn_session.close()
-            ssn = ssn.next(self._NEED_CLOSE)
+            pn_session = pn_session.next(self._REMOTE_CLOSE)
 
-        if self._pn_connection.state == (self._NEED_CLOSE):
+        if self._pn_connection.state == self._REMOTE_CLOSE:
             LOG.debug("Connection remotely closed")
-            # @todo - think about handling this wrt links!
+            self._handler.connection_remote_closed(self, None)
-            cond = self._pn_connection.remote_condition
+        elif self._pn_connection.state == self._CLOSED:
-            self._pn_connection.close()
+            LOG.debug("Connection close complete")
-            self._handler.connection_closed(self, cond)
+            self._handler.connection_closed(self)
+
+        # DEBUG LINK "LEAK"
+        # count = 0
+        # link = self._pn_connection.link_head(0)
+        # while link:
+        #     count += 1
+        #     link = link.next(0)
+        # print "Link Count %d" % count
 
         return self._next_tick
 
@@ -328,7 +413,7 @@ Associate an arbitrary user object with this Connection.
 
         pn_link = self._pn_session.sender(ident)
         if pn_link:
-            s = SenderLink(self, pn_link, ident,
+            s = SenderLink(self, pn_link,
                            source_address, target_address,
                            eventHandler, properties)
             self._sender_links[ident] = s
@@ -336,30 +421,29 @@ Associate an arbitrary user object with this Connection.
         return None
 
     def accept_sender(self, link_handle, source_override=None,
-                      event_handler=None, name=None, properties={}):
+                      event_handler=None, properties={}):
-        pn_link = self._pending_links.pop(link_handle)
+        pn_link = self._sender_links.get(link_handle)
-        if not pn_link:
+        if not pn_link or not isinstance(pn_link, proton.Sender):
             raise Exception("Invalid link_handle: %s" % link_handle)
         if pn_link.remote_source.dynamic and not source_override:
             raise Exception("A source address must be supplied!")
         source_addr = source_override or pn_link.remote_source.address
-        name = name or source_addr
+        self._sender_links[link_handle] = SenderLink(self, pn_link,
-        if name in self._sender_links:
+                                                     source_addr,
-            raise KeyError("Sender %s already exists!" % name)
+                                                     pn_link.remote_target.address,
-        self._sender_links[name] = SenderLink(self, pn_link, name,
+                                                     event_handler, properties)
-                                              source_addr,
+        return self._sender_links[link_handle]
-                                              pn_link.remote_target.address,
-                                              event_handler, properties)
-        return self._sender_links[name]
 
     def reject_sender(self, link_handle, reason):
-        pn_link = self._pending_links.pop(link_handle)
+        pn_link = self._sender_links.get(link_handle)
-        if pn_link:
+        if not pn_link or not isinstance(pn_link, proton.Sender):
-            # @todo support reason for close
+            raise Exception("Invalid link_handle: %s" % link_handle)
-            pn_link.close()
+        del self._sender_links[link_handle]
+        # @todo support reason for close
+        pn_link.close()
 
-    def create_receiver(self, target_address, source_address,
+    def create_receiver(self, target_address, source_address=None,
-                        eventHandler, name=None, properties={}):
+                        eventHandler=None, name=None, properties={}):
         """Factory for Receive links"""
         ident = name or str(target_address)
         if ident in self._receiver_links:
@@ -367,73 +451,34 @@ Associate an arbitrary user object with this Connection.
 
         pn_link = self._pn_session.receiver(ident)
         if pn_link:
-            r = ReceiverLink(self, pn_link, ident, target_address,
+            r = ReceiverLink(self, pn_link, target_address,
                              source_address, eventHandler, properties)
-            if r:
+            self._receiver_links[ident] = r
-                self._receiver_links[ident] = r
+            return r
-                return r
         return None
 
     def accept_receiver(self, link_handle, target_override=None,
-                        event_handler=None, name=None, properties={}):
+                        event_handler=None, properties={}):
-        pn_link = self._pending_links.pop(link_handle)
+        pn_link = self._receiver_links.get(link_handle)
-        if not pn_link:
+        if not pn_link or not isinstance(pn_link, proton.Receiver):
             raise Exception("Invalid link_handle: %s" % link_handle)
         if pn_link.remote_target.dynamic and not target_override:
             raise Exception("A target address must be supplied!")
         target_addr = target_override or pn_link.remote_target.address
-        name = name or target_addr
+        self._receiver_links[link_handle] = ReceiverLink(self, pn_link,
-        if name in self._receiver_links:
+                                                         target_addr,
-            raise KeyError("Receiver %s already exists!" % name)
+                                                         pn_link.remote_source.address,
-        self._receiver_links[name] = ReceiverLink(self, pn_link, name,
+                                                         event_handler, properties)
-                                                  target_addr,
+        return self._receiver_links[link_handle]
-                                              pn_link.remote_source.address,
-                                              event_handler, properties)
-        return self._receiver_links[name]
-
-        pass
 
     def reject_receiver(self, link_handle, reason):
-        pn_link = self._pending_links.pop(link_handle)
+        pn_link = self._receiver_links.get(link_handle)
-        if pn_link:
+        if not pn_link or not isinstance(pn_link, proton.Receiver):
-            # @todo support reason for close
+            raise Exception("Invalid link_handle: %s" % link_handle)
-            pn_link.close()
+        del self._receiver_links[link_handle]
+        # @todo support reason for close
+        pn_link.close()
 
-    def open(self):
-        """
-        """
-        self._pn_connection.open()
-        self._pn_session = self._pn_connection.session()
-        self._pn_session.open()
-
-    def close(self, error=None):
-        """
-        """
-        for l in self._sender_links.itervalues():
-            l.close(error)
-        for l in self._receiver_links.itervalues():
-            l.close(error)
-        self._pn_session.close()
-        self._pn_connection.close()
-
-    @property
-    def closed(self):
-        #return self._write_done and self._read_done
-        state = self._pn_connection.state
-        return state == (proton.Endpoint.LOCAL_CLOSED
-                         | proton.Endpoint.REMOTE_CLOSED)
-
-    def destroy(self):
-        """
-        """
-        self._pending_links.clear()
-        self._sender_links.clear()
-        self._receiver_links.clear()
-        self._container._remove_connection(self._name)
-        self._container = None
-        self._pn_connection = None
-        self._pn_transport = None
-        self._user_context = None
 
     def _remove_sender(self, name):
         if name in self._sender_links:

python/link.py

@@ -24,15 +24,17 @@ LOG = logging.getLogger(__name__)
 
 class _Link(object):
     """Generic Link base class"""
-    def __init__(self, connection, pn_link, name,
+    def __init__(self, connection, pn_link,
                  target_address, source_address,
                  handler, properties):
         self._connection = connection
-        self._name = name
+        self._name = pn_link.name
         self._handler = handler
         self._properties = properties
+        self._user_context = None
+        self._active = False
+        # @todo: raise jira to add 'context' attr to api
         self._pn_link = pn_link
-        # @todo: raise jira to add 'context' to api
         pn_link.context = self
 
         if target_address is None:
@@ -58,8 +60,10 @@ class _Link(object):
             else:
                 raise Exception("Unknown distribution mode: %s" %
                                 str(desired_mode))
-        self._user_context = None
+
-        self._active = False
+    @property
+    def name(self):
+        return self._name
 
     def open(self):
         """
@@ -107,18 +111,21 @@ Associate an arbitrary application object with this link.
                          | proton.Endpoint.REMOTE_CLOSED)
 
     def destroy(self):
+        LOG.debug("link destroyed %s" % str(self._pn_link))
         self._user_context = None
         self._pn_link.context = None
         self._pn_link = None
 
-
 class SenderEventHandler(object):
     """
     """
     def sender_active(self, sender_link):
         LOG.debug("sender_active (ignored)")
 
-    def sender_closed(self, sender_link, error=None):
+    def sender_remote_closed(self, sender_link, error=None):
+        LOG.debug("sender_remote_closed (ignored)")
+
+    def sender_closed(self, sender_link):
         LOG.debug("sender_closed (ignored)")
 
 
@@ -134,9 +141,9 @@ class SenderLink(_Link):
     RELEASED = 3
     MODIFIED = 4
 
-    def __init__(self, connection, pn_link, name, source_address,
+    def __init__(self, connection, pn_link, source_address,
                  target_address, eventHandler, properties):
-        super(SenderLink, self).__init__(connection, pn_link, name,
+        super(SenderLink, self).__init__(connection, pn_link,
                                          target_address, source_address,
                                          eventHandler, properties)
         self._pending_sends = collections.deque()
@@ -242,7 +249,10 @@ class ReceiverEventHandler(object):
     def receiver_active(self, receiver_link):
         LOG.debug("receiver_active (ignored)")
 
-    def receiver_closed(self, receiver_link, error=None):
+    def receiver_remote_closed(self, receiver_link, error=None):
+        LOG.debug("receiver_remote_closed (ignored)")
+
+    def receiver_closed(self, receiver_link):
         LOG.debug("receiver_closed (ignored)")
 
     def message_received(self, receiver_link, message, handle):
@@ -250,9 +260,9 @@ class ReceiverEventHandler(object):
 
 
 class ReceiverLink(_Link):
-    def __init__(self, connection, pn_link, name, target_address,
+    def __init__(self, connection, pn_link, target_address,
                  source_address, eventHandler, properties):
-        super(ReceiverLink, self).__init__(connection, pn_link, name,
+        super(ReceiverLink, self).__init__(connection, pn_link,
                                            target_address, source_address,
                                            eventHandler, properties)
         self._next_handle = 0

python/sockets.py

@@ -26,7 +26,7 @@ LOG = logging.getLogger(__name__)
     processing.
 """
 
-def read_socket_input(connection, socket):
+def read_socket_input(connection, socket_obj):
     """Read from the network layer and processes all data read.  Can
     support both blocking and non-blocking sockets.
     Returns the number of input bytes processed, or EOS if input processing
@@ -37,7 +37,7 @@ def read_socket_input(connection, socket):
         return count  # 0 or EOS
 
     try:
-        sock_data = socket.recv(count)
+        sock_data = socket_obj.recv(count)
     except socket.timeout, e:
         LOG.debug("Socket timeout exception %s", str(e))
         raise  # caller must handle
@@ -64,7 +64,7 @@ def read_socket_input(connection, socket):
         connection.close_input()
     return count
 
-def write_socket_output(connection, socket):
+def write_socket_output(connection, socket_obj):
     """Write data to the network layer.  Can support both blocking and
     non-blocking sockets.
     """
@@ -74,7 +74,7 @@ def write_socket_output(connection, socket):
 
     data = connection.output_data()
     try:
-        count = socket.send(data)
+        count = socket_obj.send(data)
     except socket.timeout, e:
         LOG.debug("Socket timeout exception %s", str(e))
         raise # caller must handle