development.ini | 1
docs/main/GettingStarted.rst | 6 ++--
docs/main/UsingLiveSocketOutsideOfMoksha.rst | 32 +++++++++++++++++++---
moksha/api/hub/consumer.py | 36 +++++++++++++++++-------
moksha/apps/helloworld/demo/consumer.py | 7 ++++
moksha/controllers/root.py | 11 ++++---
moksha/hub/amqp/__init__.py | 16 ++++++-----
moksha/hub/amqp/pyamqplib.py | 13 ++++-----
moksha/hub/amqp/qpid010.py | 2 -
moksha/hub/hub.py | 39 +++++++++++++++------------
moksha/hub/messaging.py | 2 -
moksha/lib/helpers.py | 3 ++
moksha/templates/amqp_socket.mak | 23 ++++++++++++---
moksha/templates/stomp_socket.mak | 18 +++++++-----
website/index.html | 4 +-
15 files changed, 146 insertions(+), 67 deletions(-)
New commits:
commit 3cdef7ec02045c3266ba4ae1a37af104b60aa103
Merge: a7eb71b... 23b7796...
Author: Luke Macken <lmacken(a)redhat.com>
Date: Tue Jan 18 14:15:24 2011 -0500
Merge branch 'dev'
commit 23b7796d91329e33550b45f51f921a2ef8cef486
Author: Luke Macken <lmacken(a)redhat.com>
Date: Tue Jan 18 14:03:06 2011 -0500
Fetch our AMQP javascript from the appropriate server/port/path
diff --git a/moksha/controllers/root.py b/moksha/controllers/root.py
index 83dd198..b847858 100644
--- a/moksha/controllers/root.py
+++ b/moksha/controllers/root.py
@@ -81,8 +81,6 @@ class RootController(BaseController):
data['amqp_broker_port'] = config.get('amqp_broker_port',
5672)
data['amqp_broker_user'] = config.get('amqp_broker_user',
'guest')
data['amqp_broker_pass'] = config.get('amqp_broker_pass',
'guest')
- # TODO: dynamic widget url
- # 'toscawidgets.prefix': '/toscawidgets',
data['orbited_host'] = config.get('orbited_host',
'localhost')
data['orbited_port'] = config.get('orbited_port', 9000)
@@ -90,6 +88,11 @@ class RootController(BaseController):
data['orbited_url'] = '%s://%s:%s' %
(data['orbited_scheme'],
data['orbited_host'], data['orbited_port'])
+ environ = pylons.request.environ
+ data['server'] = '%s://%s:%s%s' %
(environ.get('wsgi.url_scheme', 'http'),
+ environ['SERVER_NAME'],
+ environ['SERVER_PORT'],
+ environ['toscawidgets.prefix'])
return data
@expose('mako:moksha.templates.login')
diff --git a/moksha/templates/amqp_socket.mak b/moksha/templates/amqp_socket.mak
index f982ae3..00a73fc 100644
--- a/moksha/templates/amqp_socket.mak
+++ b/moksha/templates/amqp_socket.mak
@@ -58,9 +58,9 @@ if (typeof moksha_amqp_conn == 'undefined') {
Orbited.settings.port = ${orbited_port};
Orbited.settings.hostname = '${orbited_host}';
Orbited.settings.streaming = true;
-
$.getScript('/toscawidgets/resources/moksha.api.widgets.amqp.widgets/static/amqp.protocol.js',
function() {
-
$.getScript('/toscawidgets/resources/moksha.api.widgets.amqp.widgets/static/amqp.protocol_0_10.js',
function() {
-
$.getScript('/toscawidgets/resources/moksha.api.widgets.amqp.widgets/static/qpid_amqp.js',
function() {
+
$.getScript('${server}/resources/moksha.api.widgets.amqp.widgets/static/amqp.protocol.js',
function() {
+
$.getScript('${server}/resources/moksha.api.widgets.amqp.widgets/static/amqp.protocol_0_10.js',
function() {
+
$.getScript('${server}/resources/moksha.api.widgets.amqp.widgets/static/qpid_amqp.js',
function() {
moksha_amqp_conn = new amqp.Connection({
host: '${amqp_broker_host}',
port: ${amqp_broker_port},
commit fad10c338537385946e4cf0294b3b98280987042
Author: Luke Macken <lmacken(a)redhat.com>
Date: Tue Jan 18 14:02:35 2011 -0500
Update our "Using the live socket outside of moksha" documentation
diff --git a/docs/main/UsingLiveSocketOutsideOfMoksha.rst
b/docs/main/UsingLiveSocketOutsideOfMoksha.rst
index daa193c..f42b7f7 100644
--- a/docs/main/UsingLiveSocketOutsideOfMoksha.rst
+++ b/docs/main/UsingLiveSocketOutsideOfMoksha.rst
@@ -10,10 +10,16 @@ web page without having to write a line of Python code.
This means that you can have your web site listen to topics from your message
broker, and run javascript when new messages arrive.
+Subscribing to real-time message streams
+----------------------------------------
+
All you need to do is add a single ``<script>`` tag to your web site that
specifies the topic that you wish to listen to, and the javascript function
that should be called with each new message as they arrive.
+The Moksha live socket will automatically handle decoding the message body to
+JSON before passing it to your callback.
+
.. code-block:: html
<html>
@@ -24,14 +30,32 @@ that should be called with each new message as they arrive.
<ul id="data"/>
</body>
<script>
- function consume_message(msg) {
- $('<li/>').text(msg).appendTo('#data')
+ function consume_message(json) {
+ $('<li/>').text(json.msg).appendTo('#data')
}
</script>
<script type="text/javascript"
src="http://localhost:8080/livesocket?topic=helloworld&callback=consume_message"></script>
</html>
+
+Sending messages to the broker
+------------------------------
+
+You can easily send messages to a given topic using the ``moksha.send_message``
+function. This function will automatically handle converting your javascript
+objects to JSON for you.
+
+.. code-block:: html
+
+ <a href="#" onclick="moksha.send_message('helloworld',
{'msg': 'Hi there!'});">Send message</a>
+
+
+.. note::
+
+ In order for the live sockets to work properly, Moksha (specifically
+ Orbited) must be running under the same domain as your web site.
+
.. note::
- In order for the live sockets to work properly, Moksha (specifically Orbited) must be
running
- under the same domain as your web site.
+ You can disable the automatic JSON encoding/decoding by passing
+ ``json=False`` to the livesocket URL.
commit 208ee045e0ede9bc7b322bc89d910cc282175316
Author: Luke Macken <lmacken(a)redhat.com>
Date: Tue Jan 18 14:02:14 2011 -0500
Make the auto JSON decoding/encoding of our live socket configurable
diff --git a/moksha/controllers/root.py b/moksha/controllers/root.py
index 79eaf18..83dd198 100644
--- a/moksha/controllers/root.py
+++ b/moksha/controllers/root.py
@@ -58,7 +58,7 @@ class RootController(BaseController):
return DefaultRootController(), remainder
@expose()
- def livesocket(self, topic=None, callback=None):
+ def livesocket(self, topic=None, callback=None, json=True):
"""Returns a raw Moksha live socket, for use in non-Moksha apps.
<script> function bar(msg) { alert('bar(' + msg + ')'); }
</script>
@@ -67,7 +67,7 @@ class RootController(BaseController):
</script>
"""
- data = {'topic': topic, 'callback': callback}
+ data = {'topic': topic, 'callback': callback, 'json':
json}
backend = config.get('moksha.livesocket.backend',
'stomp').lower()
if backend == 'stomp':
override_template(self.livesocket,
'mako:moksha.templates.stomp_socket')
diff --git a/moksha/templates/amqp_socket.mak b/moksha/templates/amqp_socket.mak
index 667119f..f982ae3 100644
--- a/moksha/templates/amqp_socket.mak
+++ b/moksha/templates/amqp_socket.mak
@@ -5,7 +5,11 @@ moksha = {
accept_mode: 1,
acquire_mode: 1,
destination: 'amq.topic',
- _body: $.toJSON(body),
+% if json:
+ _body: JSON.stringify(body),
+% else:
+ _body: body,
+% endif
_header: {
delivery_properties: {
routing_key: topic
@@ -21,9 +25,18 @@ if (typeof moksha_amqp_conn == 'undefined') {
moksha_amqp_queue = null;
moksha_amqp_on_message = function(msg) {
var dest = msg.header.delivery_properties.routing_key;
+ var json = msg.body;
+% if json:
+ try {
+ var json = JSON.parse(msg.body);
+ } catch(err) {
+ console.log("Unable to decode JSON message body:");
+ console.log(f.body);
+ }
+% endif
if (moksha_callbacks[dest]) {
for (var i=0; i < moksha_callbacks[dest].length; i++) {
- moksha_callbacks[dest][i](msg.body);
+ moksha_callbacks[dest][i](json);
}
}
}
diff --git a/moksha/templates/stomp_socket.mak b/moksha/templates/stomp_socket.mak
index bcf252a..cbe1383 100644
--- a/moksha/templates/stomp_socket.mak
+++ b/moksha/templates/stomp_socket.mak
@@ -1,7 +1,11 @@
moksha = {
/* Send a STOMP message to a given topic */
send_message: function(topic, body) {
- stomp.send($.toJSON(body), topic)
+% if json:
+ stomp.send(JSON.stringify(body), topic)
+% else:
+ stomp.send(body, topic)
+% endif
},
}
@@ -41,18 +45,18 @@ if (typeof TCPSocket == 'undefined') {
};
stomp.onmessageframe = function(f){
var dest = f.headers.destination;
-/*
- var json = null;
+ var json = f.body;
+% if json:
try {
- var json = $.parseJSON(f.body);
+ var json = JSON.parse(f.body);
} catch(err) {
console.log("Unable to decode JSON message body:");
- console.log(f.body);
+ console.log(f);
}
-*/
+% endif
if (moksha_callbacks[dest]) {
for (var i=0; i < moksha_callbacks[dest].length; i++) {
- moksha_callbacks[dest][i](f.body);
+ moksha_callbacks[dest][i](json);
}
}
};
commit 585e7e9885630474824bd774781694aa5bf1af5c
Author: Luke Macken <lmacken(a)redhat.com>
Date: Tue Jan 18 11:30:06 2011 -0500
Deprecate to_unicode
diff --git a/moksha/lib/helpers.py b/moksha/lib/helpers.py
index be3bd77..365dd7d 100644
--- a/moksha/lib/helpers.py
+++ b/moksha/lib/helpers.py
@@ -812,6 +812,9 @@ def create_app_engine(app):
def to_unicode(obj, encoding='utf-8', errors='replace'):
+ """
+ :deprecated: by to_unicode in
http://python-kitchen.fedorahosted.org
+ """
if isinstance(obj, basestring):
if not isinstance(obj, unicode):
obj = unicode(obj, encoding, errors)
commit 77bce25a53d5b15b19370f18bd79eca024311739
Author: Luke Macken <lmacken(a)redhat.com>
Date: Tue Jan 18 11:29:42 2011 -0500
Make sure we can run w/o amqplib
diff --git a/moksha/hub/amqp/pyamqplib.py b/moksha/hub/amqp/pyamqplib.py
index 32c1f31..faf306f 100644
--- a/moksha/hub/amqp/pyamqplib.py
+++ b/moksha/hub/amqp/pyamqplib.py
@@ -17,7 +17,10 @@
import logging
-import amqplib.client_0_8 as amqp
+try:
+ import amqplib.client_0_8 as amqp
+except ImportError:
+ pass
from moksha.lib.helpers import trace
from moksha.hub.amqp.base import BaseAMQPHub
commit 93bac52ec4c4fc551135b39599b6a3496bede1db
Author: Luke Macken <lmacken(a)redhat.com>
Date: Tue Jan 18 11:29:22 2011 -0500
doc title changes
diff --git a/docs/main/GettingStarted.rst b/docs/main/GettingStarted.rst
index 6c08639..84cdf92 100644
--- a/docs/main/GettingStarted.rst
+++ b/docs/main/GettingStarted.rst
@@ -5,10 +5,10 @@ Getting Started with Moksha
There are currently two ways of installing and running Moksha.
.. toctree::
- :maxdepth: 1
+ :maxdepth: 0
- RPMInstallation
- VirtualenvInstallation
+ RPMInstallation (Fedora, RHEL, CentOS)
+ VirtualenvInstallation (Other linux distros, OSX, etc.)
Once you have Moksha installed, there are many ways to start using it.
commit a7eb71bcd64964f78d17cbd006b60fdae5bfe867
Author: Luke Macken <lmacken(a)redhat.com>
Date: Tue Jan 18 11:28:53 2011 -0500
More rhetoric tweaks
diff --git a/website/index.html b/website/index.html
index 1c2e6b0..a4e6475 100644
--- a/website/index.html
+++ b/website/index.html
@@ -49,7 +49,7 @@
<!-- add 2x2 grid here -->
<div class="alpha grid_5">
<h3>A Platform</h3>
- <p>Moksha is a highly-scalable platform for creating dashboards of live
widgets. Moksha makes it simple for these widgets to extract and extend data from
arbitrary resources in an intuitive and efficient manner.</p>
+ <p>Moksha is a highly-scalable platform for creating dashboards of live
widgets. Moksha makes it simple for these widgets to extract and extend data from
arbitrary resources.</p>
</div>
<div class="omega grid_5">
<h3>An Architecture</h3>
@@ -60,7 +60,7 @@
<p>Built using best-of-breed components, Moksha applications are
written using stable pre-existing modules that allow for the rapid creation of powerful
applications.</p>
</div>
<div class="omega grid_5">
- <h3>A Hub</h3>
+ <h3>A Messaging Hub</h3>
<p>Moksha acts as real-time messaging middleware for your web
application, allowing you to publish and subscribe to message queues both on server and
from the clients web browser.</p>
</div>
commit 1aa7c69fb76d66e4be2db8b0c8218972ecff1548
Author: Luke Macken <lmacken(a)redhat.com>
Date: Sun Jan 16 22:07:29 2011 -0500
log.info when the hub starts
diff --git a/moksha/hub/hub.py b/moksha/hub/hub.py
index a0c6987..4fc155c 100644
--- a/moksha/hub/hub.py
+++ b/moksha/hub/hub.py
@@ -161,6 +161,7 @@ class CentralMokshaHub(MokshaHub):
producers = None # [<Producer>,]
def __init__(self):
+ log.info('Loading the Moksha Hub')
self.topics = defaultdict(list)
self.__init_consumers()
commit de6efd68aea3f98e0cb53fe382465636d49ae418
Author: Luke Macken <lmacken(a)redhat.com>
Date: Sun Jan 16 22:04:49 2011 -0500
Update our hub to use more up to date terminology with variable/method names
diff --git a/moksha/hub/hub.py b/moksha/hub/hub.py
index 5371474..a0c6987 100644
--- a/moksha/hub/hub.py
+++ b/moksha/hub/hub.py
@@ -158,7 +158,7 @@ class CentralMokshaHub(MokshaHub):
The Moksha Hub is responsible for initializing all of the Hooks,
AMQP queues, exchanges, etc.
"""
- data_streams = None # [<DataStream>,]
+ producers = None # [<Producer>,]
def __init__(self):
self.topics = defaultdict(list)
@@ -170,9 +170,11 @@ class CentralMokshaHub(MokshaHub):
self.__init_amqp()
self.__run_consumers()
- self.__init_data_streams()
+ self.__init_producers()
def __init_amqp(self):
+ # Ok this looks odd at first. I think this is only used when we are briding
stomp/amqp,
+ # Since each producer and consumer opens up their own AMQP connections anyway
if self.stomp_broker:
log.debug("Initializing local AMQP queue...")
self.server_queue_name = 'moksha_hub_' + self.session.name
@@ -187,7 +189,7 @@ class CentralMokshaHub(MokshaHub):
def __init_consumers(self):
""" Initialize all Moksha Consumer objects """
- log.info('Loading Moksha Consumers')
+ log.info('Loading Consumers')
for consumer in pkg_resources.iter_entry_points('moksha.consumer'):
c_class = consumer.load()
log.info("%s consumer is watching the %r topic" % (
@@ -203,15 +205,16 @@ class CentralMokshaHub(MokshaHub):
self.consumers.append(c)
self.topics[topic][i] = c.consume
- def __init_data_streams(self):
- """ Initialize all data streams """
- self.data_streams = []
+ def __init_producers(self):
+ """ Initialize all producers (aka data streams)
"""
+ log.info('Loading Producers')
+ self.producers = []
for entry in ('moksha.producer', 'moksha.stream'):
- for stream in pkg_resources.iter_entry_points(entry):
- stream_class = stream.load()
- log.info('Loading %s producer' % stream_class.__name__)
- stream_obj = stream_class()
- self.data_streams.append(stream_obj)
+ for producer in pkg_resources.iter_entry_points(entry):
+ producer_class = producer.load()
+ log.info('Loading %s producer' % producer_class.__name__)
+ producer_obj = producer_class()
+ self.producers.append(producer_obj)
@trace
def create_topic(self, topic):
@@ -225,10 +228,10 @@ class CentralMokshaHub(MokshaHub):
def stop(self):
log.debug("Stopping the CentralMokshaHub")
MokshaHub.close(self)
- if self.data_streams:
- for stream in self.data_streams:
- log.debug("Stopping data stream %s" % stream)
- stream.stop()
+ if self.producers:
+ for producer in self.producers:
+ log.debug("Stopping producer %s" % producer)
+ producer.stop()
if self.consumers:
for consumer in self.consumers:
log.debug("Stopping consumer %s" % consumer)
commit d842655d335dd690e6ba954a8e17e65a9824ce99
Author: Luke Macken <lmacken(a)redhat.com>
Date: Sun Jan 16 21:56:06 2011 -0500
Get rid of moksha.streams
diff --git a/moksha/streams/__init__.py b/moksha/streams/__init__.py
deleted file mode 100644
index e69de29..0000000
commit 0ea2e2c3d207837cc894996fefdc7ba9cf6cdb2e
Author: Luke Macken <lmacken(a)redhat.com>
Date: Sat Jan 15 15:12:25 2011 -0500
Commit our db changes in our HelloWorld Consumer every 100 entries.
diff --git a/moksha/apps/helloworld/demo/consumer.py
b/moksha/apps/helloworld/demo/consumer.py
index 10f4e3b..dec4cfd 100644
--- a/moksha/apps/helloworld/demo/consumer.py
+++ b/moksha/apps/helloworld/demo/consumer.py
@@ -4,6 +4,7 @@ from demo.model import HelloWorldModel
class HelloWorldConsumer(Consumer):
topic = 'helloworld'
app = 'helloworld'
+ i = 0
def consume(self, message):
self.log.info('Received message: ' +
message['body']['msg'])
@@ -11,4 +12,8 @@ class HelloWorldConsumer(Consumer):
entry = HelloWorldModel()
entry.message = message['body']['msg']
self.DBSession.add(entry)
- self.DBSession.commit()
+
+ self.i += 1
+ if self.i % 100 == 0:
+ self.DBSession.commit()
+
commit 92145193e14be58e8d36b19a4cbacc065d0cd8c9
Author: Luke Macken <lmacken(a)redhat.com>
Date: Wed Jan 12 17:39:35 2011 -0500
Some method renaming in our AMQP stack to prevent conflicts with STOMP
diff --git a/moksha/api/hub/consumer.py b/moksha/api/hub/consumer.py
index 1c5712c..0d24094 100644
--- a/moksha/api/hub/consumer.py
+++ b/moksha/api/hub/consumer.py
@@ -54,9 +54,9 @@ class Consumer(object):
self.hub.queue_declare(queue=queue_name, exclusive=True)
self.hub.exchange_bind(queue_name, binding_key=topic)
if self.jsonify:
- self.hub.subscribe(queue_name, self._consume_json)
+ self.hub.queue_subscribe(queue_name, self._consume_json)
else:
- self.hub.subscribe(queue_name, self._consume)
+ self.hub.queue_subscribe(queue_name, self._consume)
else:
# Assume we're using Qpid then.
server_queue_name = 'moksha_consumer_' +
self.hub.session.name
diff --git a/moksha/hub/amqp/pyamqplib.py b/moksha/hub/amqp/pyamqplib.py
index e6a031f..32c1f31 100644
--- a/moksha/hub/amqp/pyamqplib.py
+++ b/moksha/hub/amqp/pyamqplib.py
@@ -77,13 +77,12 @@ class AMQPLibHub(BaseAMQPHub):
msg = self.channel.basic_get(queue, no_ack=True)
return msg
- def subscribe(self, queue, callback, no_ack=True):
+ def queue_subscribe(self, queue, callback, no_ack=True):
"""
Consume messages from a given `queue`, passing each to `callback`
"""
self.channel.basic_consume(queue, callback=callback, no_ack=no_ack)
-
def wait(self):
self.channel.wait()
commit e89f2a4aa5a70402e1887fcb3bf4fe6883434a47
Author: Luke Macken <lmacken(a)redhat.com>
Date: Wed Jan 12 17:06:09 2011 -0500
Remove a tracing decorator, which may be causing problems
diff --git a/moksha/hub/amqp/pyamqplib.py b/moksha/hub/amqp/pyamqplib.py
index b9ab46c..e6a031f 100644
--- a/moksha/hub/amqp/pyamqplib.py
+++ b/moksha/hub/amqp/pyamqplib.py
@@ -77,7 +77,6 @@ class AMQPLibHub(BaseAMQPHub):
msg = self.channel.basic_get(queue, no_ack=True)
return msg
- @trace
def subscribe(self, queue, callback, no_ack=True):
"""
Consume messages from a given `queue`, passing each to `callback`
commit fc36b9bc2970c611a4fb5063463f27cfd96df21d
Author: Luke Macken <lmacken(a)redhat.com>
Date: Wed Jan 12 17:00:40 2011 -0500
Update our MessagingHub.subscribe method arguments
diff --git a/moksha/hub/messaging.py b/moksha/hub/messaging.py
index dc6fc08..56eb782 100644
--- a/moksha/hub/messaging.py
+++ b/moksha/hub/messaging.py
@@ -25,5 +25,5 @@ class MessagingHub(object):
def send_message(self, topic, message):
raise NotImplementedError
- def subscribe(self, topic):
+ def subscribe(self, topic, callback):
raise NotImplementedError
commit 349423f0cb112ae2ba663ad3c217897d462ef503
Author: Luke Macken <lmacken(a)redhat.com>
Date: Fri Jan 7 20:14:15 2011 -0500
Typo
diff --git a/moksha/api/hub/consumer.py b/moksha/api/hub/consumer.py
index 5ce870f..1c5712c 100644
--- a/moksha/api/hub/consumer.py
+++ b/moksha/api/hub/consumer.py
@@ -25,6 +25,7 @@ loaded, and receives each message for the specified topic through the
.. moduleauthor:: Luke Macken <lmacken(a)redhat.com>
"""
+import uuid
import logging
log = logging.getLogger('moksha.hub')
@@ -49,8 +50,7 @@ class Consumer(object):
if isinstance(self.hub, AMQPLibHub):
# AMQPLibHub specific
- import uuid
- queue_name = str(uuid.uiud4())
+ queue_name = str(uuid.uuid4())
self.hub.queue_declare(queue=queue_name, exclusive=True)
self.hub.exchange_bind(queue_name, binding_key=topic)
if self.jsonify:
commit 057ea1b0ef3457f3af04ce1a3b9ff6df86b6a4e6
Author: Luke Macken <lmacken(a)redhat.com>
Date: Fri Jan 7 20:12:48 2011 -0500
Temporary hack around our leaking amqphub abstraction
diff --git a/moksha/api/hub/consumer.py b/moksha/api/hub/consumer.py
index 83b528f..5ce870f 100644
--- a/moksha/api/hub/consumer.py
+++ b/moksha/api/hub/consumer.py
@@ -29,6 +29,7 @@ import logging
log = logging.getLogger('moksha.hub')
from moksha.hub.hub import MokshaHub
+from moksha.hub.amqp.pyamqplib import AMQPLibHub
from moksha.lib.helpers import listify, create_app_engine, json
from sqlalchemy.orm import sessionmaker
@@ -45,18 +46,31 @@ class Consumer(object):
if self.hub.amqp_broker and not self.hub.stomp_broker:
for topic in listify(self.topic):
log.debug('Subscribing to consumer topic %s' % topic)
- server_queue_name = 'moksha_consumer_' + self.hub.session.name
- self.hub.queue_declare(queue=server_queue_name, exclusive=True)
- self.hub.exchange_bind(server_queue_name, binding_key=topic)
- local_queue_name = 'moksha_consumer_' + self.hub.session.name
- self.hub.local_queue = self.hub.session.incoming(local_queue_name)
- self.hub.message_subscribe(queue=server_queue_name,
- destination=local_queue_name)
- self.hub.local_queue.start()
- if self.jsonify:
- self.hub.local_queue.listen(self._consume_json)
+
+ if isinstance(self.hub, AMQPLibHub):
+ # AMQPLibHub specific
+ import uuid
+ queue_name = str(uuid.uiud4())
+ self.hub.queue_declare(queue=queue_name, exclusive=True)
+ self.hub.exchange_bind(queue_name, binding_key=topic)
+ if self.jsonify:
+ self.hub.subscribe(queue_name, self._consume_json)
+ else:
+ self.hub.subscribe(queue_name, self._consume)
else:
- self.hub.local_queue.listen(self._consume)
+ # Assume we're using Qpid then.
+ server_queue_name = 'moksha_consumer_' +
self.hub.session.name
+ self.hub.queue_declare(queue=server_queue_name, exclusive=True)
+ self.hub.exchange_bind(server_queue_name, binding_key=topic)
+ local_queue_name = 'moksha_consumer_' +
self.hub.session.name
+ self.hub.local_queue = self.hub.session.incoming(local_queue_name)
+ self.hub.message_subscribe(queue=server_queue_name,
+ destination=local_queue_name)
+ self.hub.local_queue.start()
+ if self.jsonify:
+ self.hub.local_queue.listen(self._consume_json)
+ else:
+ self.hub.local_queue.listen(self._consume)
# If the consumer specifies an 'app', then setup `self.engine` to
# be a SQLAlchemy engine, along with a configured DBSession
commit 66056971ccd26c619630f33030706befd62b60e2
Author: Luke Macken <lmacken(a)redhat.com>
Date: Fri Jan 7 17:03:41 2011 -0500
Pass some more AMQP broker configuration through to the pyamqplib layer
diff --git a/development.ini b/development.ini
index 7b938e8..0ff234a 100644
--- a/development.ini
+++ b/development.ini
@@ -105,6 +105,7 @@ stomp_pass = guest
#amqp_broker_port = 5672
#amqp_broker_user = guest
#amqp_broker_pass = guest
+#amqp_broker_ssl = False
# Documentation directory
docs_dir = docs
diff --git a/moksha/hub/amqp/pyamqplib.py b/moksha/hub/amqp/pyamqplib.py
index fc0f5e4..b9ab46c 100644
--- a/moksha/hub/amqp/pyamqplib.py
+++ b/moksha/hub/amqp/pyamqplib.py
@@ -29,8 +29,8 @@ NONPERSISTENT_DELIVERY = PERSISTENT_DELIVERY = range(1, 3)
class AMQPLibHub(BaseAMQPHub):
""" An AMQPHub implemention using the amqplib module
"""
- def __init__(self, broker, username=None, password=None, ssl=False):
- self.conn = amqp.Connection(host=broker, ssl=ssl,
+ def __init__(self, broker, username=None, password=None, ssl=False, threaded=False):
+ self.conn = amqp.Connection(host=broker, ssl=ssl, use_threading=threaded,
userid=username, password=password)
self.channel = self.conn.channel()
self.channel.access_request('/data', active=True, write=True, read=True)
diff --git a/moksha/hub/amqp/qpid010.py b/moksha/hub/amqp/qpid010.py
index 980c67e..0460420 100644
--- a/moksha/hub/amqp/qpid010.py
+++ b/moksha/hub/amqp/qpid010.py
@@ -35,7 +35,7 @@ class QpidAMQPHub(BaseAMQPHub):
"""
- def __init__(self, broker):
+ def __init__(self, broker, **kw):
self.set_broker(broker)
self.socket = connect(self.host, self.port)
if self.url.scheme == URL.AMQPS:
diff --git a/moksha/hub/hub.py b/moksha/hub/hub.py
index 819f098..5371474 100644
--- a/moksha/hub/hub.py
+++ b/moksha/hub/hub.py
@@ -72,7 +72,10 @@ class MokshaHub(StompHub, AMQPHub):
self.topics[topic].append(callback)
if self.amqp_broker:
- AMQPHub.__init__(self, self.amqp_broker)
+ AMQPHub.__init__(self, self.amqp_broker,
+ username=config.get('amqp_broker_user',
'guest'),
+ password=config.get('amqp_broker_pass',
'guest'),
+ ssl=config.get('amqp_broker_ssl', False))
if self.stomp_broker:
log.info('Initializing STOMP support')
commit a23bea017d53dc56adad9edcaacee97236667237
Author: Luke Macken <lmacken(a)redhat.com>
Date: Fri Jan 7 16:19:25 2011 -0500
Try to fall back to pyamqplib if qpid doesn't exist
diff --git a/moksha/hub/amqp/__init__.py b/moksha/hub/amqp/__init__.py
index 7ab1ec9..98ada27 100644
--- a/moksha/hub/amqp/__init__.py
+++ b/moksha/hub/amqp/__init__.py
@@ -23,10 +23,14 @@ try:
from qpid010 import QpidAMQPHub
AMQPHub = QpidAMQPHub
except ImportError:
- log.debug("Unable to import qpid module")
- class FakeHub(object):
- pass
- AMQPHub = FakeHub
+ log.debug("Cannot find qpid python module")
+ try:
+ from pyamqplib import AMQPLibHub
+ AMQPHub = AMQPLibHub
+ except ImportError:
+ log.debug("Cannot find pyamqplib")
+ log.debug("Using FakeHub AMQP broker. Don't expect AMQP to work")
+ class FakeHub(object):
+ pass
+ AMQPHub = FakeHub
-#from pyamqplib import AMQPLibHub
-#AMQPHub = AMQPLibHub