Branch 'amqp' - 10 commits - development.ini moksha/api moksha/hub moksha/lib orbited.cfg
by Luke Macken
development.ini | 2
moksha/api/hub/consumer.py | 17 ++++++++
moksha/api/widgets/amqp/amqp.py | 6 --
moksha/api/widgets/live/live.py | 18 ++++----
moksha/hub/amqp/qpid010.py | 4 -
moksha/hub/hub.py | 83 +++++++++++++++++++---------------------
moksha/lib/helpers.py | 4 +
orbited.cfg | 2
8 files changed, 77 insertions(+), 59 deletions(-)
New commits:
commit ed6e3748f51b671c944799a426d15af18f9fa0df
Author: Luke Macken <lmacken(a)redhat.com>
Date: Wed Nov 11 22:39:27 2009 -0500
Disable STOMP support by default
diff --git a/development.ini b/development.ini
index b3d18ca..3d0fa9a 100644
--- a/development.ini
+++ b/development.ini
@@ -85,7 +85,7 @@ orbited_host = localhost
orbited_port = 9000
# Stomp broker configuration.
-stomp_broker = localhost
+#stomp_broker = localhost
stomp_port = 61613
stomp_user = guest
stomp_pass = guest
diff --git a/orbited.cfg b/orbited.cfg
index b7c2195..cbabfe7 100644
--- a/orbited.cfg
+++ b/orbited.cfg
@@ -4,7 +4,7 @@ http://:9000
# This enables Orbited's built-in MorbidQ message queue.
# Comment this out if to use your own, eg: RabbitMQ
-stomp://:61613
+#stomp://:61613
[access]
* -> localhost:61613
commit 7817d5b0dc06c60ca82ada8173641b9756662e74
Author: Luke Macken <lmacken(a)redhat.com>
Date: Wed Nov 11 22:38:27 2009 -0500
[hub] only setup the local/remote AMQP queue in the hub if stomp is
enabled.
diff --git a/moksha/hub/hub.py b/moksha/hub/hub.py
index 138c4d3..0e097b3 100644
--- a/moksha/hub/hub.py
+++ b/moksha/hub/hub.py
@@ -120,16 +120,17 @@ class CentralMokshaHub(MokshaHub):
self.__init_data_streams()
def __init_amqp(self):
- log.debug("Initializing local AMQP queue...")
- self.server_queue_name = 'moksha_hub_' + self.session.name
- self.queue_declare(queue=self.server_queue_name, exclusive=True)
- self.exchange_bind(self.server_queue_name, binding_key='#')
- self.local_queue_name = 'moksha_hub'
- self.local_queue = self.session.incoming(self.local_queue_name)
- self.message_subscribe(queue=self.server_queue_name,
- destination=self.local_queue_name)
- self.local_queue.start()
- self.local_queue.listen(self.consume_amqp_message)
+ if self.stomp_broker:
+ log.debug("Initializing local AMQP queue...")
+ self.server_queue_name = 'moksha_hub_' + self.session.name
+ self.queue_declare(queue=self.server_queue_name, exclusive=True)
+ self.exchange_bind(self.server_queue_name, binding_key='#')
+ self.local_queue_name = 'moksha_hub'
+ self.local_queue = self.session.incoming(self.local_queue_name)
+ self.message_subscribe(queue=self.server_queue_name,
+ destination=self.local_queue_name)
+ self.local_queue.start()
+ self.local_queue.listen(self.consume_amqp_message)
def __init_consumers(self):
""" Initialize all Moksha Consumer objects """
commit 75d670b1f44d87b11ae8da706b6f41a54bee9d3c
Author: Luke Macken <lmacken(a)redhat.com>
Date: Wed Nov 11 22:37:16 2009 -0500
[consumer] If we're using AMQP w/o STOMP, have the consumers setup
their own remote & local queues with their MokshaHub.
diff --git a/moksha/api/hub/consumer.py b/moksha/api/hub/consumer.py
index ac2495f..c3f9ed2 100644
--- a/moksha/api/hub/consumer.py
+++ b/moksha/api/hub/consumer.py
@@ -26,7 +26,11 @@ loaded, and receives each message for the specified topic through the
.. moduleauthor:: Luke Macken <lmacken(a)redhat.com>
"""
+import logging
+log = logging.getLogger('moksha.hub')
+
from moksha.hub.hub import MokshaHub
+from moksha.lib.helpers import listify
class Consumer(object):
""" A message consumer """
@@ -34,6 +38,19 @@ class Consumer(object):
def __init__(self):
self.hub = MokshaHub()
+ self.log = log
+ 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()
+ self.hub.local_queue.listen(self.consume)
def consume(self, message):
raise NotImplementedError
commit 6ec6fabf8fc4fe63e94ff49246ed6463cab5bee2
Author: Luke Macken <lmacken(a)redhat.com>
Date: Wed Nov 11 22:36:44 2009 -0500
Add a moksha.lib.helpers.listify method
diff --git a/moksha/lib/helpers.py b/moksha/lib/helpers.py
index 45aa72b..4995921 100644
--- a/moksha/lib/helpers.py
+++ b/moksha/lib/helpers.py
@@ -1039,3 +1039,7 @@ def get_num_cpus():
def deprecation(message):
warnings.warn(message, DeprecationWarning)
+
+
+def listify(something):
+ return not isinstance(something, list) and [something] or something
commit 5d8e0d2c30454713694fbeda7a7d1af3325e1f1d
Author: Luke Macken <lmacken(a)redhat.com>
Date: Wed Nov 11 22:36:32 2009 -0500
Minor module cleanup
diff --git a/moksha/hub/hub.py b/moksha/hub/hub.py
index 6d5e203..138c4d3 100644
--- a/moksha/hub/hub.py
+++ b/moksha/hub/hub.py
@@ -18,7 +18,6 @@
from moksha.hub.reactor import reactor
-import os
import sys
import signal
import pkg_resources
@@ -26,7 +25,6 @@ import logging
from tg import config
from orbited import json
-from threading import Thread
from paste.deploy import appconfig
from moksha.lib.helpers import trace, defaultdict, get_moksha_config_path
commit c2918583b4d118afcd81b15a4104146d6020f0e1
Author: Luke Macken <lmacken(a)redhat.com>
Date: Wed Nov 11 22:35:44 2009 -0500
Move the consume_{amqp,stomp}_message methods out of the MokshaHub and
into the CentralMokshaHub class.
diff --git a/moksha/hub/hub.py b/moksha/hub/hub.py
index 862ed0e..6d5e203 100644
--- a/moksha/hub/hub.py
+++ b/moksha/hub/hub.py
@@ -96,41 +96,12 @@ class MokshaHub(StompHub, AMQPHub):
This method will cause the specified `callback` to be executed with
each message that goes through a given topic.
"""
+ log.debug('watch_topic(%s)' % locals())
if len(self.topics[topic]) == 0:
if self.stomp_broker:
self.subscribe(topic)
self.topics[topic].append(callback)
- def consume_amqp_message(self, message):
- self.message_accept(message)
- topic = message.headers[0]['routing_key']
- try:
- body = json.decode(message.body)
- except Exception, e:
- log.warning('Cannot decode message from JSON: %s' % e)
- body = message.body
- if self.stomp_broker:
- StompHub.send_message(self, topic.encode('utf8'),
- message.body.encode('utf8'))
-
- def consume_stomp_message(self, message):
- topic = message['headers'].get('destination')
- if not topic:
- return
-
- # We can enable this if/when we need it...
- #try:
- # body = json.decode(message['body'])
- #except Exception, e:
- # log.warning('Cannot decode message from JSON: %s' % e)
- # body = message['body']
-
-
- # feed all of our consumers
- for callback in self.topics.get(topic, []):
- reactor.callInThread(callback, message)
-
-
class CentralMokshaHub(MokshaHub):
"""
The Moksha Hub is responsible for initializing all of the Hooks,
@@ -154,7 +125,7 @@ class CentralMokshaHub(MokshaHub):
log.debug("Initializing local AMQP queue...")
self.server_queue_name = 'moksha_hub_' + self.session.name
self.queue_declare(queue=self.server_queue_name, exclusive=True)
- self.exchange_bind(self.server_queue_name)
+ self.exchange_bind(self.server_queue_name, binding_key='#')
self.local_queue_name = 'moksha_hub'
self.local_queue = self.session.incoming(self.local_queue_name)
self.message_subscribe(queue=self.server_queue_name,
@@ -204,6 +175,35 @@ class CentralMokshaHub(MokshaHub):
log.debug("Stopping data stream %s" % stream)
stream.stop()
+ def consume_amqp_message(self, message):
+ self.message_accept(message)
+ topic = message.headers[0]['routing_key']
+ try:
+ body = json.decode(message.body)
+ except Exception, e:
+ log.warning('Cannot decode message from JSON: %s' % e)
+ log.debug('Message: %r' % message.body)
+ body = message.body
+ if self.stomp_broker:
+ StompHub.send_message(self, topic.encode('utf8'),
+ message.body.encode('utf8'))
+
+ def consume_stomp_message(self, message):
+ topic = message['headers'].get('destination')
+ if not topic:
+ return
+
+ # We can enable this if/when we need it...
+ #try:
+ # body = json.decode(message['body'])
+ #except Exception, e:
+ # log.warning('Cannot decode message from JSON: %s' % e)
+ # body = message['body']
+
+ # feed all of our consumers
+ for callback in self.topics.get(topic, []):
+ reactor.callInThread(callback, message)
+
def setup_logger(verbose):
global log
commit 4593e855177f8909166a65b9395947110996bfce
Author: Luke Macken <lmacken(a)redhat.com>
Date: Wed Nov 11 19:29:12 2009 -0500
Fix a bug in our Qpid hub
diff --git a/moksha/hub/amqp/qpid010.py b/moksha/hub/amqp/qpid010.py
index 0ed5e5d..2af134d 100644
--- a/moksha/hub/amqp/qpid010.py
+++ b/moksha/hub/amqp/qpid010.py
@@ -16,11 +16,9 @@
#
# Authors: Luke Macken <lmacken(a)redhat.com>
-import qpid
import logging
from qpid.util import connect, URL, ssl
-from qpid.queue import Empty
from qpid.datatypes import Message, uuid4, RangedSet
from qpid.connection import Connection
@@ -66,7 +64,7 @@ class QpidAMQPHub(BaseAMQPHub):
msg = Message(props, message)
self.session.message_transfer(destination=exchange, message=msg)
- def subscribe_queue(server_queue_name, local_queue_name):
+ def subscribe_queue(self, server_queue_name, local_queue_name):
queue = self.session.incoming(local_queue_name)
self.session.message_subscribe(queue=server_queue_name,
destination=local_queue_name)
commit a14d8edd2c000547deca66309cc950327552c2d1
Author: Luke Macken <lmacken(a)redhat.com>
Date: Wed Nov 11 19:28:16 2009 -0500
More AMQPSocket/LiveWidget integration
diff --git a/moksha/api/widgets/amqp/amqp.py b/moksha/api/widgets/amqp/amqp.py
index 14a8ebc..2259e92 100644
--- a/moksha/api/widgets/amqp/amqp.py
+++ b/moksha/api/widgets/amqp/amqp.py
@@ -74,14 +74,11 @@ class AMQPSocket(Widget):
'orbited_port', 'orbited_url', 'orbited_js', 'amqp_broker_host',
'amqp_broker_port', 'amqp_broker_user', 'amqp_broker_pass',
'send_hook', 'recieve_hook']
- onconnectedframe = amqp_subscribe('org.fedoraproject.#')
+ onconnectedframe = ''
onmessageframe = ''
send_hook = ''
recieve_hook = ''
- # Popup notification bubbles on socket state changes
- notify = False
-
engine_name = 'mako'
template = u"""
<script type="text/javascript">
diff --git a/moksha/api/widgets/live/live.py b/moksha/api/widgets/live/live.py
index 706402c..74608e8 100644
--- a/moksha/api/widgets/live/live.py
+++ b/moksha/api/widgets/live/live.py
@@ -23,7 +23,7 @@ from tw.api import Widget
from moksha.exc import MokshaException
from moksha.api.widgets.stomp import StompWidget, stomp_subscribe, stomp_unsubscribe
-from moksha.api.widgets.amqp import amqp_subscribe, amqp_unsubscribe
+from moksha.api.widgets.amqp import amqp_subscribe, amqp_unsubscribe, AMQPSocket
class LiveWidget(Widget):
""" A live streaming widget.
@@ -39,14 +39,11 @@ class LiveWidget(Widget):
template = 'mako:myproject.templates.mylivewidget'
"""
- callbacks = ['onmessage']
engine_name = 'mako'
def __init__(self, id, *args, **kw):
super(LiveWidget, self).__init__(*args, **kw)
- backend = tg.config.get('moksha.livesocket.backend', 'amqp').lower()
- if backend == 'stomp':
- self.callbacks.extend(StompWidget.callbacks)
+ self.backend = tg.config.get('moksha.livesocket.backend', 'amqp').lower()
def update_params(self, d):
""" Register this widgets message topic callbacks """
@@ -56,7 +53,12 @@ class LiveWidget(Widget):
if not topics:
raise MokshaException('You must specify a `topic` to subscribe to')
topics = isinstance(topics, list) and topics or [topics]
- for callback in self.callbacks:
+ callbacks = []
+ if self.backend == 'stomp':
+ callbacks = StompWidget.callbacks
+ elif self.backend == 'amqp':
+ callbacks = AMQPSocket.callbacks
+ for callback in callbacks:
if callback == 'onmessageframe':
for topic in topics:
cb = getattr(self, 'onmessage').replace('${id}', self.id)
commit d42c0fbceac6fa85520bbc382810f2e8e9aa1bb2
Author: Luke Macken <lmacken(a)redhat.com>
Date: Wed Nov 11 19:27:08 2009 -0500
Make sure we return something from the unsubscribe_method
diff --git a/moksha/api/widgets/amqp/amqp.py b/moksha/api/widgets/amqp/amqp.py
index 989913c..14a8ebc 100644
--- a/moksha/api/widgets/amqp/amqp.py
+++ b/moksha/api/widgets/amqp/amqp.py
@@ -57,6 +57,7 @@ def amqp_unsubscribe(topic):
""" Return a javascript callback that unsubscribes to a given topic,
or a list of topics.
"""
+ return ""
# TODO:
#sub = "stomp.unsubscribe('%s');"
#if isinstance(topic, list):
commit cf7015f5cf55c71824f136dfa4d9fa0cf0ff9389
Author: Luke Macken <lmacken(a)redhat.com>
Date: Wed Nov 11 14:30:59 2009 -0500
Yes, class methods take classes as a first arg
diff --git a/moksha/api/widgets/live/live.py b/moksha/api/widgets/live/live.py
index ab360b9..706402c 100644
--- a/moksha/api/widgets/live/live.py
+++ b/moksha/api/widgets/live/live.py
@@ -80,7 +80,7 @@ class LiveWidget(Widget):
return topics
@classmethod
- def subscribe_topics(topics):
+ def subscribe_topics(cls, topics):
backend = tg.config.get('moksha.livesocket.backend', 'stomp').lower()
if backend == 'amqp':
return amqp_subscribe(topics)
@@ -92,7 +92,7 @@ class LiveWidget(Widget):
"'stomp'." % backend)
@classmethod
- def unsubscribe_topics(topics):
+ def unsubscribe_topics(cls, topics):
backend = tg.config.get('moksha.livesocket.backend', 'stomp').lower()
if backend == 'amqp':
return amqp_unsubscribe(topics)
14 years, 6 months
Branch 'amqp' - 2 commits - moksha/api moksha/apps moksha/tests
by Luke Macken
moksha/api/widgets/amqp/amqp.py | 5 +----
moksha/apps/demo/MokshaJQPlotDemo/jqplotdemo/tests/functional/test_root.py | 2 +-
moksha/tests/functional/test_root.py | 2 +-
moksha/tests/quickstarts/test_livewidget.py | 2 +-
4 files changed, 4 insertions(+), 7 deletions(-)
New commits:
commit 463b62a9fdc10ba5eabd20e22ade8fdd0cc04fe2
Author: Luke Macken <lmacken(a)redhat.com>
Date: Wed Nov 11 14:26:01 2009 -0500
Update our livewidget tests to handle the new AMQPSocket
diff --git a/moksha/apps/demo/MokshaJQPlotDemo/jqplotdemo/tests/functional/test_root.py b/moksha/apps/demo/MokshaJQPlotDemo/jqplotdemo/tests/functional/test_root.py
index b8d4950..525a336 100644
--- a/moksha/apps/demo/MokshaJQPlotDemo/jqplotdemo/tests/functional/test_root.py
+++ b/moksha/apps/demo/MokshaJQPlotDemo/jqplotdemo/tests/functional/test_root.py
@@ -24,7 +24,7 @@ class TestRootController(TestController):
def test_tcpsocket(self):
""" Make sure the Moksha socket gets injected """
response = self.app.get('/')
- assert_true('TCPSocket' in response)
+ assert_true('TCPSocket' in response or 'moksha_amqp_conn' in response)
def test_jqplot(self):
""" Make sure jQuery is injected properly """
diff --git a/moksha/tests/functional/test_root.py b/moksha/tests/functional/test_root.py
index 370af70..a1527ee 100644
--- a/moksha/tests/functional/test_root.py
+++ b/moksha/tests/functional/test_root.py
@@ -54,4 +54,4 @@ class TestRootController(TestController):
def test_tcpsocket(self):
""" Ensure our TCP socket is getting injected """
resp = self.app.get('/')
- assert 'TCPSocket' in resp
+ assert 'TCPSocket' in resp or 'moksha_amqp_conn' in resp
diff --git a/moksha/tests/quickstarts/test_livewidget.py b/moksha/tests/quickstarts/test_livewidget.py
index 1cdcf98..72a505b 100644
--- a/moksha/tests/quickstarts/test_livewidget.py
+++ b/moksha/tests/quickstarts/test_livewidget.py
@@ -59,7 +59,7 @@ class TestLiveWidgetQuickstart(QuickstartTester):
def test_tcpsocket(self):
""" Ensure our Orbited TCPSocket is getting injected """
resp = self.app.get('/')
- assert 'TCPSocket' in resp
+ assert 'TCPSocket' in resp or 'moksha_amqp_conn' in resp
def test_livewidget(self):
""" Ensure our LiveWidget is available """
commit 628e319c3a09f7d992e8720b468cbdb2ba8c21f9
Author: Luke Macken <lmacken(a)redhat.com>
Date: Wed Nov 11 14:25:50 2009 -0500
Remove some spurious console.logs
diff --git a/moksha/api/widgets/amqp/amqp.py b/moksha/api/widgets/amqp/amqp.py
index 6d0a68c..989913c 100644
--- a/moksha/api/widgets/amqp/amqp.py
+++ b/moksha/api/widgets/amqp/amqp.py
@@ -90,7 +90,7 @@ class AMQPSocket(Widget):
moksha_amqp_remote_queue = null;
moksha_amqp_queue = null;
moksha_amqp_on_message = function(msg) {
- console.log('moksha_amqp_on_message,(' + msg.body + ')');
+ console.log('moksha_amqp_on_message(' + msg.body + ')');
var dest = msg.header.delivery_properties.routing_key;
var json = $.secureEvalJSON(msg.body);
if (moksha_callbacks[dest]) {
@@ -120,7 +120,6 @@ class AMQPSocket(Widget):
Orbited.settings.port = ${orbited_port};
Orbited.settings.hostname = '${orbited_host}';
Orbited.settings.streaming = true;
- console.log('Opening AMQP connection');
moksha_amqp_conn = new amqp.Connection({
host: '${amqp_broker_host}',
port: ${amqp_broker_port},
@@ -131,14 +130,12 @@ class AMQPSocket(Widget):
});
moksha_amqp_conn.start();
- console.log('Starting AMQP session');
moksha_amqp_session = moksha_amqp_conn.create_session(
'moksha_socket_' + (new Date().getTime() + Math.random()));
moksha_amqp_remote_queue = 'moksha_socket_queue_' +
moksha_amqp_session.name;
- console.log('Declaring AMQP queue');
moksha_amqp_session.Queue('declare', {
queue: moksha_amqp_remote_queue
});
14 years, 6 months
Changes to 'amqp'
by Luke Macken
New branch 'amqp' available with the following commits:
commit 726a5084e7b0e1e71becf2f6213341f6eece3a4b
Author: Luke Macken <lmacken(a)redhat.com>
Date: Wed Nov 11 13:13:05 2009 -0500
Make our LiveWidget more protocol-agnostic
commit f948b369c54d6c3ee8e9e7c8516b497afd0a9691
Author: Luke Macken <lmacken(a)redhat.com>
Date: Wed Nov 11 13:12:51 2009 -0500
More AMQPSocket cleanups
commit 666f7099be518155f2c55e7c82c6a666e8182322
Author: Luke Macken <lmacken(a)redhat.com>
Date: Wed Nov 11 13:03:12 2009 -0500
Support proxying to AMQP by default with Orbited :)
commit 146e2eeba43c4123dc2efc28442402632166d82d
Author: Luke Macken <lmacken(a)redhat.com>
Date: Wed Nov 11 12:34:37 2009 -0500
Add the {,un}subscribe_topic methods as classmethods on our socket
commit 1a2bc5f7436efc25fca9a741569edc3c1f8f2fcf
Author: Luke Macken <lmacken(a)redhat.com>
Date: Wed Nov 11 12:33:48 2009 -0500
Let kamaloka deal with the TCPSocket, and refactor some of of our AMQPSocket
commit 2f360a4a4e6e8548fddf6af20ef06974ef5acb5e
Author: Luke Macken <lmacken(a)redhat.com>
Date: Wed Nov 11 12:32:46 2009 -0500
Tweak some of our AMQPSocket callbacks, and define some in the LiveWidget
commit 1c0792082545e1ebf90a4ff61738d51380d8f132
Author: Luke Macken <lmacken(a)redhat.com>
Date: Wed Nov 11 12:31:59 2009 -0500
We only need to start the AMQP queue once
commit a18dbac3b86b1469f7e290f97510d78092f51488
Author: Luke Macken <lmacken(a)redhat.com>
Date: Wed Nov 11 12:31:11 2009 -0500
Refator the `moksha.stomp` to `moksha.livewidget` to be protocol-agnostic.
commit 07741cecdd72b74560cbb150acbc26d2e043b327
Author: Luke Macken <lmacken(a)redhat.com>
Date: Tue Nov 10 21:40:38 2009 -0500
Initial commit of our AMQP-driven live widgets.
This creates a new AMQPSocket widget, based on the StompWidget, which
utilizes the kamaloka AMQP javascript bindings. At this moment in time
the widget can connect back up to Qpid, through Orbited, and can subscribe
to message queues.
14 years, 6 months
moksha.spec setup.py
by John (J5) Palmieri
moksha.spec | 2 +-
setup.py | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
New commits:
commit fe04db1b30fd42da7968f9883b81b47e55a971e2
Author: John (J5) Palmieri <johnp(a)redhat.com>
Date: Mon Nov 2 14:48:07 2009 -0500
new version
diff --git a/moksha.spec b/moksha.spec
index 245c981..b78f877 100644
--- a/moksha.spec
+++ b/moksha.spec
@@ -2,7 +2,7 @@
%{!?pyver: %define pyver %(%{__python} -c "import sys ; print sys.version[:3]")}
Name: moksha
-Version: 0.4.2
+Version: 0.4.3
Release: 1%{?dist}
Summary: A flexable platform for creating live collaborative web applications
Group: Applications/Internet
diff --git a/setup.py b/setup.py
index 90469dd..2bfb452 100644
--- a/setup.py
+++ b/setup.py
@@ -19,7 +19,7 @@ from paver.setuputils import find_package_data
setup(
name='moksha',
- version='0.4.2',
+ version='0.4.3',
description='',
author='',
author_email='',
14 years, 6 months