Pull in changes from bitbucket to make sure it's up to date.

Author: benjaminws

TODO

@@ -1 +1 @@
-* Get better performance from get_line and parse_frame?
+* Be more awesome

docs/reference/index.rst

@@ -6,7 +6,7 @@
 :Date: |today|
 
 .. toctree::
-    :maxdepth: 3
+    :maxdepth: 4
 
     stompy.stomp
     stompy.frame 

docs/reference/stompy.simple.rst

@@ -0,0 +1,10 @@
+=====================================
+Simple Client - stompy.simple
+=====================================
+
+.. currentmodule:: stompy.simple
+
+.. automodule:: stompy.simple
+    :members:
+
+

examples/simple_example.py

@@ -5,12 +5,13 @@
 from itertools import count
 from optparse import OptionParser
 
+def callback_func(message):
+    print("Using Callback Function")
+    print(message.headers.get('message-id'))
+    print(message.body)
+    stomp.ack(message)
 
-def consume(host, port, queue, num=None):
-    stomp = StompClient(host, port)
-    # optional connect keyword args "username" and "password" like so:
-    # stomp.connect(username="user", password="pass")
-    stomp.connect()
+def consume(queue, num=None, callback=None):
 
     stomp.subscribe(queue, ack="client")
 
@@ -24,17 +25,16 @@ def _handle_message(frame):
 
     try:
         for i in it:
-            frame = stomp.get()
-            _handle_message(frame)
+            if callback:
+                stomp.get(callback=callback_func)
+            else:
+                frame = stomp.get()
+                _handle_message(frame)
     finally:
         stomp.disconnect()
 
 
-def produce(host, port, queue, num=1000):
-    stomp = StompClient(host, port)
-    # optional connect keyword args "username" and "password" like so:
-    # stomp.connect(username="user", password="pass")
-    stomp.connect()
+def produce(queue, num=1000):
 
     for i in xrange(0, num):
         print("Message #%d" % i)
@@ -58,6 +58,9 @@ def produce(host, port, queue, num=1000):
                       default=False, dest='produce', help='produce messages')
     parser.add_option('-c', '--consume', action='store_true',
                       default=False, dest='consume', help='consume messages')
+    parser.add_option('-C', '--use-callback', action='store_true',
+                      default=False, dest='callback',
+                      help='send retrieved message to python callable')
     parser.add_option('-n', '--number', action='store',
                       type='int', dest='number', default="100",
                       help='produce or consume NUMBER messages')
@@ -69,8 +72,13 @@ def produce(host, port, queue, num=1000):
         parser.print_help()
         sys.exit(1)
 
+    stomp = StompClient(options.host, options.port)
+    # optional connect keyword args "username" and "password" like so:
+    # stomp.connect(username="user", password="pass")
+    stomp.connect()
+
     if options.produce:
         print("PRODUCING")
-        produce(options.host, options.port, options.queue, options.number)
+        produce(options.queue, options.number)
     elif options.consume:
-        consume(options.host, options.port, options.queue, options.number)
+        consume(options.queue, options.number, options.callback)

examples/stomp_example.py

@@ -1,58 +1,44 @@
 #!/usr/bin/env python
 import sys
+import itertools
 import time
 from stompy import Stomp
 from optparse import OptionParser
 
+def callback_func(message):
+    print("Using Callback Function")
+    print(message.headers.get('message-id'))
+    print(message.body)
+    stomp.ack(message)
 
-def consume(host, port, queue, num=None):
-    try:
-        stomp = Stomp(host, port)
-        # optional connect keyword args "username" and "password" like so:
-        # stomp.connect(username="user", password="pass")
-        stomp.connect()
-    except:
-        print("Cannot connect")
-        raise
+def consume(queue, num=None, callback=None):
 
     # If using RabbitMQ, the queue seems to 'disappear' 
     # after disconnecting a consumer, to make the queue persistent
     # add the headers 'auto-delete': 'false' and 'durable': 'true'
     # to the dictionary below
     stomp.subscribe({'destination': queue, 'ack': 'client'})
 
-    if not num:
-        while True:
-            try:
-                frame = stomp.receive_frame()
-                stomp.ack(frame)
-                print(frame.headers.get('message-id'))
-                print(frame.body)
-            except KeyboardInterrupt:
-                stomp.disconnect()
-                break
-    else:
-        for i in xrange(0, num):
-            try:
+    def _handle_message(frame):
+        print(frame.headers.get('message-id'))
+        print(frame.body)
+        stomp.ack(frame)
+
+    # if num is not set, iterate forever.
+    it = xrange(0, num) if num else itertools.count()
+
+    try:
+        for i in it:
+            if callback:
+                stomp.receive_frame(callback=callback_func)
+            else:
                 frame = stomp.receive_frame()
-                stomp.ack(frame)
-                print(frame.headers.get('message-id'))
-                print(frame.body)
-            except KeyboardInterrupt:
-                stomp.disconnect()
-                break
+                _handle_message(frame)
+    finally:
         stomp.disconnect()
 
 
-def produce(host, port, queue, num=1000):
-    try:
-        stomp = Stomp(host, port)
-        # optional connect keyword args "username" and "password" like so:
-        # stomp.connect(username="user", password="pass")
-        stomp.connect()
-    except:
-        print("Cannot connect")
-        raise
+def produce(queue, num=1000):
 
     for i in xrange(0, num):
         print("Message #%d" % i)
@@ -75,6 +61,9 @@ def produce(host, port, queue, num=1000):
                       default=False, dest='produce', help='produce messages')
     parser.add_option('-c', '--consume', action='store_true',
                       default=False, dest='consume', help='consume messages')
+    parser.add_option('-C', '--use-callback', action='store_true',
+                      default=False, dest='callback',
+                      help='send retrieved message to python callable')
     parser.add_option('-n', '--number', action='store',
                       type='int', dest='number',
                       help='produce or consume NUMBER messages')
@@ -94,7 +83,16 @@ def produce(host, port, queue, num=1000):
         parser.print_help()
         sys.exit(1)
 
+    try:
+        stomp = Stomp(options.host, options.port)
+        # optional connect keyword args "username" and "password" like so:
+        # stomp.connect(username="user", password="pass")
+        stomp.connect()
+    except:
+        print("Cannot connect")
+        raise
+
     if options.produce:
-        produce(options.host, options.port, options.queue, options.number)
+        produce(options.queue, options.number)
     elif options.consume:
-        consume(options.host, options.port, options.queue, options.number)
+        consume(options.queue, options.number, options.callback)

setup.cfg

@@ -2,6 +2,8 @@
 source-dir = docs/
 build-dir = docs/build
 all_files = 1
+[upload_sphinx]
+upload-dir = docs/build/html
 [nosetests]
 verbosity = 1
 detailed-errors = 1

setup.py

@@ -12,7 +12,6 @@
       url=distmeta.__homepage__,
       keywords='stomp activemq jms messaging',
       test_suite="nose.collector",
-      setup_requires=['nose>=0.11', 'dingus'],
       classifiers=["Development Status :: 5 - Production/Stable",
                    "Intended Audience :: Developers",
                    "License :: OSI Approved :: BSD License",

stompy/distmeta.py

@@ -1,6 +1,6 @@
 """Implementation of the STOMP protocol in Python.
 """
-VERSION = (0, 2, 4)
+VERSION = (0, 2, 9)
 __version__ = ".".join(map(str, VERSION))
 __author__ = "Benjamin W. Smith"
 __contact__ = "benjaminwarfield@just-another.net"

stompy/frame.py

@@ -1,7 +1,7 @@
 import socket
 import random
 from pprint import pformat
-from errno import EAGAIN
+from errno import EAGAIN, EWOULDBLOCK
 from Queue import Queue
 from Queue import Empty as QueueEmpty
 
@@ -66,21 +66,28 @@ def __init__(self, sock=None):
         self.iqueue = IntermediateMessageQueue()
         self.rqueue = Queue()
 
-    def connect(self, sock, username=None, password=None):
+    def connect(self, sock, username=None, password=None, clientid=None):
         """Connect to the STOMP server and get the session id.
 
         :param sock: Socket object from stompy.stomp.Stomp.
         :keyword username: Username for connection.
         :keyword password: Password for connection.
+        :keyword clientid: Client identification for persistent connections
 
         """
         self.sock = sock
+
+        headers = {}
+
         if username and password:
-            frame = self.build_frame({"command": "CONNECT",
-                                      "headers": {'login': username,
-                                                  'passcode': password}})
-        else:
-            frame = self.build_frame({"command": "CONNECT", "headers": {}})
+            headers.update({'login': username,
+                           'passcode': password})
+	
+        if clientid:        
+            headers.update({'client-id' : clientid})
+
+                    
+        frame = self.build_frame({"command": "CONNECT", "headers": headers})
 
         self.send_frame(frame.as_string())
 
@@ -259,10 +266,12 @@ def _getline(self, nb=False):
             while not buffer.endswith('\x00'):
                 try:
                     partial = self.sock.recv(1)
+                    if not partial or partial == '':
+                        raise UnknownBrokerResponseError('empty reply')
                 except socket.error, exc:
-                    if exc[0] == EAGAIN:
+                    if exc[0] == EAGAIN or exc[0] == EWOULDBLOCK:
                         if not buffer or buffer == '\n':
-                            return None
+                            raise UnknownBrokerResponseError('empty reply')
                         continue
                 buffer += partial
         finally:

stompy/simple.py

@@ -19,7 +19,7 @@ class Client(object):
         >>> from stompy.simple import Client
         >>> stomp = Client()
         >>> stomp.connect()
-        >>> stomp.send("The quick brown fox...", destination="/queue/test")
+        >>> stomp.put("The quick brown fox...", destination="/queue/test")
         >>> stomp.subscribe("/queue/test")
         >>> message = stomp.get_nowait()
         >>> message.body
@@ -35,17 +35,19 @@ def __init__(self, host="localhost", port=61613):
         self.stomp = Stomp(host, port)
         self._current_transaction = None
 
-    def get(self, block=True):
+    def get(self, block=True, callback=None):
         """Get message.
 
         :keyword block: Block if necessary until an item is available.
             If this is ``False``, return an item if one is immediately
             available, else raise the :exc:`Empty` exception.
 
+        :keyword callback: Optional function to execute when message recieved.
+
         :raises Empty: If ``block`` is off and no message was receied.
 
         """
-        frame = self.stomp.receive_frame(nonblocking=not block)
+        frame = self.stomp.receive_frame(nonblocking=not block, callback=callback)
         if frame is None and not block:
             raise self.Empty()
         return frame
@@ -78,19 +80,20 @@ def put(self, item, destination, persistent=True, conf=None):
 
         return self.stomp.send(conf)
 
-    def connect(self, username=None, password=None):
+    def connect(self, username=None, password=None, clientid=None):
         """Connect to the broker.
 
         :keyword username: Username for connection
         :keyword password: Password for connection
+        :keyword clientid: Client identification for persistent connections
 
         :raises :exc:`stompy.stomp.ConnectionError`:
             if the connection was unsuccessful.
         :raises :exc:`stompy.stomp.ConnectionTimeoutError`:
             if the connection timed out.
 
         """
-        self.stomp.connect(username=username, password=password)
+        self.stomp.connect(username=username, password=password, clientid=clientid)
 
     def disconnect(self):
         """Disconnect from the broker."""