Part of Slepp's Projects — Pastebin — TURL — Imagebin — Filebin
Feedback -- English French German Japanese
Create Upload Newest Tools Donate
Sign In | Create Account

Paste Actions

Download Raw
Embed
pBox URL
Post to Twitter
Remove Paste

Paste Details

Python Source format, 1 year 8 months old, and 857 hits.

Paste Description for Plugging AMQP and WebSockets

see http://www.defuze.org/archives/207-plugging-amqp-and-websocketplugging-amqp-and-websocket.html

Plugging AMQP and WebSockets
Saturday, June 12th, 2010 at 11:45:38am MDT 

  1. # -*- coding: utf-8 -*-
  2. import sys
  3. import time
  4. import select
  5. import logging
  6. from logging import handlers
  7. if hasattr(select, "poll"):
  8.     from asyncore import poll2 as poll
  9. else:
  10.     from asyncore import poll
  11.  
  12. import pika
  13. import tornado
  14. import tornado.httpserver
  15. import tornado.ioloop
  16. from tornado import websocket
  17. from cherrypy.process import wspbus, plugins
  18.  
  19. class MyBus(wspbus.Bus):
  20.     def __init__(self, name=""):
  21.         wspbus.Bus.__init__(self)
  22.         self.open_logger(name)
  23.         self.subscribe("log", self._log)
  24.  
  25.         self.ioloop = tornado.ioloop.IOLoop.instance()
  26.         self.ioloop.add_callback(self.call_main)
  27.  
  28.     def call_main(self):
  29.         self.publish('main')
  30.         time.sleep(0.1)
  31.         self.ioloop.add_callback(self.call_main)
  32.        
  33.     def block(self):
  34.         ioloop = tornado.ioloop.IOLoop.instance()
  35.         try:
  36.             ioloop.start()
  37.         except KeyboardInterrupt:
  38.             ioloop.stop()
  39.             self.exit()
  40.  
  41.     def exit(self):
  42.         wspbus.Bus.exit(self)
  43.         self.close_logger()
  44.  
  45.     def open_logger(self, name=""):
  46.         logger = logging.getLogger(name)
  47.         logger.setLevel(logging.INFO)
  48.         h = logging.StreamHandler(sys.stdout)
  49.         h.setLevel(logging.INFO)
  50.         h.setFormatter(logging.Formatter("[%(asctime)s] %(name)s - %(levelname)s - %(message)s"))
  51.         logger.addHandler(h)
  52.  
  53.         self.logger = logger
  54.  
  55.     def close_logger(self):
  56.         for handler in self.logger.handlers:
  57.             handler.flush()
  58.             handler.close()
  59.            
  60.     def _log(self, msg="", level=logging.INFO):
  61.         self.logger.log(level, msg)
  62.  
  63. class WS2AMQPPlugin(plugins.SimplePlugin):
  64.     def __init__(self, bus):
  65.         plugins.SimplePlugin.__init__(self, bus)
  66.         self.conn = pika.AsyncoreConnection(pika.ConnectionParameters('localhost'))
  67.         self.channel = self.conn.channel()
  68.         self.channel.exchange_declare(exchange="X", type="direct", durable=False)
  69.         self.channel.queue_declare(queue="Q", durable=False, exclusive=False)
  70.         self.channel.queue_bind(queue="Q", exchange="X", routing_key="")
  71.  
  72.         self.channel.basic_consume(self.amqp2ws, queue="Q")
  73.  
  74.         self.bus.subscribe("ws2amqp", self.ws2amqp)
  75.         self.bus.subscribe("stop", self.cleanup)
  76.  
  77.     def cleanup(self):
  78.         self.bus.unsubscribe("ws2amqp", self.ws2amqp)
  79.         self.bus.unsubscribe("stop", self.cleanup)
  80.         self.channel.queue_delete(queue="Q")
  81.         self.channel.exchange_delete(exchange="X")
  82.         self.conn.close()
  83.  
  84.     def amqp2ws(self, ch, method, header, body):
  85.         self.bus.publish("amqp2ws", body)
  86.         ch.basic_ack(delivery_tag=method.delivery_tag)
  87.        
  88.     def ws2amqp(self, message):
  89.         self.bus.log("Publishing to AMQP: %s" % message)
  90.         self.channel.basic_publish(exchange="X", routing_key="", body=message)
  91.        
  92. bus = MyBus()
  93.  
  94. class WebSocket2AMQP(websocket.WebSocketHandler):
  95.     def __init__(self, *args, **kwargs):
  96.         websocket.WebSocketHandler.__init__(self, *args, **kwargs)
  97.         self.settings['bus'].subscribe("amqp2ws", self.push_message)
  98.        
  99.     def open(self):
  100.         self.receive_message(self.on_message)
  101.  
  102.     def on_message(self, message):
  103.         self.settings['bus'].publish("ws2amqp", message)
  104.         self.write_message(message)
  105.         self.receive_message(self.on_message)
  106.  
  107.     def on_connection_close(self):
  108.         self.settings['bus'].unsubscribe("amqp2ws", self.push_message)
  109.  
  110.     def push_message(self, message):
  111.         self.write_message(message)
  112.  
  113. if __name__ == '__main__':
  114.     application = tornado.web.Application([
  115.         (r"/ws", WebSocket2AMQP),
  116.         ], bus=bus)
  117.    
  118.     http_server = tornado.httpserver.HTTPServer(application)
  119.     http_server.listen(8888)
  120.    
  121.     bus.subscribe("main", poll)
  122.     WS2AMQPPlugin(bus).subscribe()
  123.     bus.start()
  124.     bus.block()

Paste Details

Tags: python amqp websockets

advertising

Update the Post

Either update this post and resubmit it with changes, or make a new post.

You may also comment on this post.

update paste below
details of the post (optional)

Note: Only the paste content is required, though the following information can be useful to others.

Save name / title?

(space separated, optional)



Please note that information posted here will expire by default in one month. If you do not want it to expire, please set the expiry time above. If it is set to expire, web search engines will not be allowed to index it prior to it expiring. Items that are not marked to expire will be indexable by search engines. Be careful with your passwords. All illegal activities will be reported and any information will be handed over to the authorities, so be good.

worth-right