- implemented connection, session, and link endpoints

- added a simple I/O driver
- added a simple client library
- implemented a simple broker along with send and recv utilities

README

@@ -2,19 +2,62 @@ This project contains a partial prototype of AMQP 1-0 PR2. This
 currently includes relatively complete components for framing and
 codec.
 
+Getting Started:
+
+  1. First run the broker, passing the names of your queues on the
+     command line:
+
+       ./broker queue-a queue-b queue-c ...
+
+  2. Run the send program to send a message:
+
+       ./send queue-a this is a test
+
+  3. Run the recv program to receive a message:
+
+       ./recv queue-b
+
+  4. Use the --help option for each of the above programs to explore
+     in more depth. All of the above programs support the -t/--trace
+     option which may be used to print a protocol trace:
+
+       ./send -t raw queue-a trace raw bytes
+       ./send -t ops queue-b trace operations before/after encode/decode
+       ./send -t "raw ops err" queue-c trace everything
+
 Files:
 
+  broker         -- A prototype broker: ./broker --help
+
+  recv           -- A client used to receive messages: ./recv --help
+
+  send           -- A client used to send messages: ./send --help
+
+  client.py      -- A simple client library.
+
   codec.py       -- An implementation of the AMQP type system.
 
+  concurrency.py -- Concurrency utilities.
+
+  connection.py  -- An implementation of an AMQP connection endpoint.
+
   framing.py     -- An implementation of the AMQP framing layer.
 
+  link.py        -- An implementation of an AMQP link endpoint.
+
   mllib/         -- An XML parsing library used to load type definitions.
 
   operations.py  -- Classes representing all the operations and related
                     types defined by the AMQP transport specification.
 
   README         -- This file.
 
+  queue.py       -- A simple AMQP queue implementation.
+
+  selector.py    -- A simple I/O driver.
+
+  session.py
+
   test.py        -- Some informal test code.
 
   transport.xml  -- Definitions from the AMQP transport specification.

broker

@@ -0,0 +1,144 @@
+#!/usr/bin/python
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+import optparse, socket
+from connection import Connection
+from session import Session, SessionError
+from link import LinkError, Receiver, Sender, link
+from selector import Acceptor, Selector
+from queue import Queue, DESTRUCTIVE
+from util import ConnectionSelectable
+
+class Broker:
+
+  def __init__(self):
+    self.queues = {}
+    self.cursors = {}
+    self.init = {Sender.direction: self.init_sender,
+                 Receiver.direction: self.init_receiver}
+    self.process = {Sender.direction:self.process_sender,
+                    Receiver.direction: self.process_receiver}
+
+  def tick(self, connection):
+    if connection.opening():
+      connection.open()
+
+    for ssn in connection.sessions.values():
+      if ssn.attaching():
+        ssn.attach()
+
+      for link in ssn.links.values():
+        if link.opening():
+          if link.local in self.queues:
+            link.link()
+            self.init[link.direction](link)
+          else:
+            link.local = None
+            link.link()
+            link.unlink()
+            ssn.remove(link)
+            continue
+
+        self.process[link.direction](link)
+
+        if link.closing():
+          link.unlink()
+          ssn.remove(link)
+
+      if ssn.detaching():
+        ssn.detach(True)
+        connection.remove(ssn)
+
+    if connection.closing():
+      connection.close()
+
+    connection.tick()
+
+  def init_sender(self, link):
+    q = self.queues[link.source]
+    cursor = q.cursor(DESTRUCTIVE)
+    self.cursors[link] = cursor, {}
+
+  def init_receiver(self, link):
+    link.flow(10)
+
+  def process_sender(self, link):
+    cursor, unacked = self.cursors[link]
+    while link.capacity() > 0:
+      entry = cursor.get()
+      if entry is None:
+        break
+      elif entry.item is None:
+        continue
+      else:
+        xfr = entry.item
+        print "TRANSFER:", xfr
+        tag = link.send(fragments = xfr.fragments)
+        unacked[tag] = entry
+    while link.pending():
+      t, d = link.get()
+      e = unacked[t]
+      xfr = e.item
+      e.dequeue()
+      print "DEQUEUED:", xfr
+      del unacked[t]
+      link.settle(t)
+
+  def process_receiver(self, link):
+    while link.pending():
+      xfr = link.get()
+      self.queues[link.target].put(xfr)
+      print "ENQUEUED:", xfr
+      link.ack(xfr)
+    if link.capacity() < 10: link.flow(10)
+
+def bind(host, port):
+  sock = socket.socket()
+  sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+  sock.bind((host, port))
+  sock.listen(5)
+  sock.setblocking(0)
+  return sock
+
+parser = optparse.OptionParser(usage="usage: %prog [options] QUEUE_1 ... QUEUE_n",
+                               description="Prototype amqp broker.")
+parser.add_option("-i", "--interface", default="0.0.0.0",
+                  help="interface to listen on (default %default)")
+parser.add_option("-p", "--port", type=int, default=5672,
+                  help="port to listen on (default %default)")
+parser.add_option("-t", "--trace", default="err",
+                  help="enable tracing for specified categories")
+
+opts, args = parser.parse_args()
+
+broker = Broker()
+
+for a in args:
+  broker.queues[a] = Queue()
+
+selector = Selector()
+sock = bind(opts.interface, opts.port)
+
+def handler(accepted):
+  conn = Connection(lambda n: Session(n, link))
+  conn.tracing = set(opts.trace.split())
+  selector.register(ConnectionSelectable(accepted, conn, broker.tick))
+
+selector.register(Acceptor(sock, handler))
+selector.run()

client.py

@@ -0,0 +1,136 @@
+#!/usr/bin/python
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+import socket
+from connection import Connection as BaseConnection
+from session import Session as BaseSession, SessionError
+from link import Sender as BaseSender, Receiver as BaseReceiver, LinkError, link
+from selector import Selector
+from util import ConnectionSelectable
+from concurrency import synchronized, Condition, Waiter
+from threading import RLock
+from uuid import uuid4
+from operations import Fragment
+
+class Connection(BaseConnection):
+
+  def __init__(self):
+    BaseConnection.__init__(self, self.session)
+    self._lock = RLock()
+    self.condition = Condition(self._lock)
+    self.waiter = Waiter(self.condition)
+    self.selector = Selector.default()
+
+  def connect(self, host, port):
+    sock = socket.socket()
+    sock.connect((host, port))
+    sock.setblocking(0)
+    self.selector.register(ConnectionSelectable(sock, self, self._tick))
+
+  @synchronized
+  def _tick(self, connection):
+    connection.tick()
+    self.waiter.notify()
+
+  def wait(self, predicate, timeout=10):
+    self.selector.wakeup()
+    self.waiter.wait(predicate, timeout)
+
+  def session(self, name = None):
+    if name is None:
+      name = str(uuid4())
+    ssn = Session(self, name)
+    self.add(ssn)
+    ssn.attach()
+    return ssn
+
+  @synchronized
+  def close(self):
+    BaseConnection.close(self)
+    self.wait(lambda: self.close_rcvd)
+
+class Session(BaseSession):
+
+  def __init__(self, connection, name):
+    BaseSession.__init__(self, name, link)
+    self.connection = connection
+    self._lock = self.connection._lock
+
+  def wait(self, predicate, timeout=10):
+    self.connection.wait(predicate, timeout)
+
+  @synchronized
+  def sender(self, target):
+    snd = Sender(self.connection, target)
+    self.add(snd)
+    snd.link()
+    self.wait(lambda: snd.opened())
+    if snd.target is None:
+      snd.close()
+      raise LinkError("no such target: %s" % target)
+    return snd
+
+  @synchronized
+  def receiver(self, source, limit=0):
+    rcv = Receiver(self.connection, source)
+    self.add(rcv)
+    rcv.link()
+    if limit:
+      rcv.flow(limit)
+    self.wait(lambda: rcv.opened())
+    if rcv.source is None:
+      rcv.close()
+      raise LinkError("no such source: %s" % source)
+    return rcv
+
+  @synchronized
+  def close(self):
+    self.detach(True)
+
+class Link:
+
+  def __init__(self, connection):
+    self.connection = connection
+    self._lock = self.connection._lock
+
+  def wait(self, predicate, timeout=10):
+    self.connection.wait(predicate, timeout)
+
+  @synchronized
+  def close(self):
+    self.unlink()
+    self.wait(lambda: self.closed())
+
+class Sender(BaseSender, Link):
+
+  def __init__(self, connection, target):
+    BaseSender.__init__(self, str(uuid4()), None, target)
+    Link.__init__(self, connection)
+
+  @synchronized
+  def send(self, **kwargs):
+    self.wait(self.capacity)
+    return BaseSender.send(self, **kwargs)
+
+class Receiver(BaseReceiver, Link):
+
+  def __init__(self, connection, source):
+    BaseReceiver.__init__(self, str(uuid4()), source, None)
+    Link.__init__(self, connection)

codec.py

@@ -104,6 +104,7 @@ def __init__(self, encodings=ENCODINGS):
       str: self.enc_string,
       Symbol: self.enc_symbol,
       buffer: self.enc_binary,
+      uuid.UUID: self.enc_uuid,
       None.__class__: self.enc_null
       }
     self.deconstructors = {