first working version

Author: tbarbugli

.gitignore

@@ -0,0 +1,28 @@
+*.py[cod]
+
+# C extensions
+*.so
+
+# Packages
+*.egg
+*.egg-info
+dist
+build
+eggs
+parts
+bin
+var
+sdist
+develop-eggs
+.installed.cfg
+lib
+lib64
+__pycache__
+
+# Installer logs
+pip-log.txt
+
+# Unit test / coverage reports
+.coverage
+.tox
+nosetests.xml

README.md

@@ -0,0 +1,147 @@
+Tyron
+======
+Tyron is a web app to send push events to connected clients
+
+A client connects to tyron and subscribes to a specific channel,
+tyron will keep the http connection open until something get published
+to that channel.
+
+To keep the connection count as little as possible open connection will
+timeout after a configurable interval of time.
+
+The messaging backend utilises the redis pub/sub feature.
+
+Subscribe to channels
+====================
+Clients subscribes to channel using the /<channel>/ entry point
+
+Publish to clients
+==================
+To publish to a client you need to connect to redis and publish
+on the channel tyron subscribes (see CHANNEL_NAME setting)
+
+NOTE: The user channel is part of the message you publish to redis
+
+Message format
+==============
+Messages must follow this json format:
+
+`{"channel": "user_channel", "data": "...."}`
+
+
+Example:
+========
+
+client side: POST to /hiphop/
+tyron subscribes the client to hiphop channel
+
+`{"channel": "hiphop", "data": "...."}`
+
+
+Install
+=======
+
+Just do pip install tyron or do a setup.py install
+
+Note: If you are using OSX you might need something like this:
+
+`CFLAGS="-I /opt/local/include -L /opt/local/lib" pip install tyron`
+
+Settings
+======== 
+
+**DEBUG**
+
+Run Flask in debug mode
+
+**LONGPOLLING_TIMEOUT**
+
+`Default: 3`
+
+The timeout in seconds for connections waiting for 
+
+**TIMEOUT\_RESPONSE_MESSAGE**
+
+`Default: str('')`
+
+The message the webserver will return to clients when LONGPOLLING_TIMEOUT occours
+
+**CHANNEL_NAME**
+
+`Default: tyron_pubsub`
+
+The redis pub/sub channel tyron subscribes to
+
+**REDIS_HOSTNAME**
+
+`Default: localhost`
+
+
+**REDIS_PORT**
+
+`Default: 6379`
+
+
+**REDIS_DB**
+
+`Default: 0`
+
+
+**LOG_LEVEL**
+
+`Default: logging.INFO`
+
+The loglevel (from python logging module)
+
+You can override settings pointing tyron to your own settings module.
+The settings module must be specified in the TYRON_CONF environment variable.
+
+eg.
+
+`TYRON_CONF=my_settings.py python tyron.py`
+
+
+DEPLOY
+======
+
+tyron is a WSGI app so you have lots of choice in terms of web servers :)
+You definetely wants to opt for some async worker as tyron will keep an high amount of open idle connections (so gevent/libevent ...)
+
+I personally like the semplicity of green unicorn:
+
+
+`
+pip install gunicorn
+gunicorn -b :8091 -w 9 -k gevent --worker-connections=2000 tyron --log-level=critical
+`   
+
+few notes about chosen parameters:
+
+-k gevent tells gunicorn to use the gevent worker
+-w 9 means, spawn 9 worker processes (2-4 * cpu_cores is suggested)
+--worker-connections 2000 every worker process will handle 2000 connections
+
+
+DEPLOY CAVEATS
+==============
+
+On high traffic web sites is quite easy to hit some common limit of your kernel
+limits, the first limit you are probably going to hit is the file descriptor limit.
+By default Ubuntu (and many other OS) have the very low file descriptor limit of 1024
+I suggest to benchmark your machine running tyron to find the best value for this limit and update it.
+
+Verbose logging can lead to issues, unless the logging facility you are using handles that for you.
+Logging INFO/DEBUG messages can easily lead to high cpu usage.
+
+
+BENCHMARKS
+==========
+
+coming soon ...
+
+.using an EC2 medium instance and bees with machine guns
+
+
+
+
+

setup.py

@@ -0,0 +1,51 @@
+from setuptools import find_packages
+from setuptools import setup
+
+description="""
+Gevent redis/pubsub event notifier written in flask and gevent
+"""
+
+setup(
+    name="tyron",
+    version='0.0.2',
+    url='https://github.com/tbarbugli/tyron',
+    license='BSD',
+    platforms=['OS Independent'],
+    description = description.strip(),
+    author = 'Tommaso Barbugli',
+    author_email = '[email protected]',
+    maintainer = 'Tommaso Barbugli',
+    maintainer_email = '[email protected]',
+    packages=find_packages(),
+    tests_require=[
+        'mock',
+    ],
+    test_suite='tests',
+    dependency_links = [
+        'https://github.com/downloads/SiteSupport/gevent/gevent-1.0rc2.tar.gz#egg=gevent'
+    ],
+    requires = [
+        'Flask (==0.9)',
+        'Jinja2 (==2.6)',
+        'Werkzeug (==0.8.3)',
+        'greenlet (==0.4.0)',
+        'gunicorn (==0.17.2)',
+        'redis (==2.7.2)',
+        'ujson (==1.30)',
+    ],
+    entry_points={
+        'console_scripts': [
+            'tyron = tyron:main',
+        ],
+    },
+    classifiers=[
+        'Development Status :: 4 - Beta',
+        'Framework :: Django',
+        'Intended Audience :: Developers',
+        'License :: OSI Approved :: BSD License',
+        'Operating System :: OS Independent',
+        'Programming Language :: Python',
+        'Programming Language :: Python :: 2.7',
+        'Topic :: Internet :: WWW/HTTP',
+    ]
+)

tests.py

@@ -0,0 +1,43 @@
+import json
+import mock
+import redis
+from redis.client import PubSub
+from mock import MagicMock
+from mock import patch
+import unittest
+from tyron.tyron import subscriptions
+from tyron.tyron import RedisSub
+
+redis_pubsub = RedisSub('channel', 'localhost', 6379, 1)
+
+class TestRedisPubSub(unittest.TestCase):
+
+    @patch.object(PubSub, 'listen')
+    @patch.object(PubSub, 'execute_command')
+    def test_subscribes_to_redis(self, pubsub_exec, *args):
+        redis_pubsub.subscribe()
+        pubsub_exec.assert_called_with('SUBSCRIBE', 'channel')
+
+    def test_redis_connection_from_init(self):
+        redis_pubsub = RedisSub('channel', 'redis.local', 6380, 1, 'xxx')
+        client = redis_pubsub.get_redis_connection()
+        configs = client.connection_pool.connection_kwargs
+        assert configs['host'] == 'redis.local'
+        assert configs['port'] == 6380
+        assert configs['db'] == 1
+        assert configs['password'] == 'xxx'
+
+    def test_message_parsing(self):
+        message = {
+            'channel': 'channelname',
+            'data': 'data'
+        }
+        encoded_message = json.dumps(message)
+        r_message = redis_pubsub.parse_message(encoded_message)
+        assert 'channelname', 'data' == r_message
+
+    def test_redis_config(self):
+        pass
+
+    def test_timeout(self):
+        pass

tyron/__init__.py

@@ -0,0 +1,11 @@
+import logging
+
+DEBUG = False
+TIMEOUT_RESPONSE_MESSAGE = ''
+LONGPOLLING_TIMEOUT = 15
+CHANNEL_NAME = 'notifications_pubsub'
+REDIS_HOSTNAME = 'localhost'
+REDIS_PORT = 6379
+REDIS_DB = 0
+HTTP_PORT = 8080
+LOG_LEVEL = logging.INFO

tyron/default_settings.py

@@ -0,0 +1,93 @@
+from collections import defaultdict
+
+try:
+    import ujson as json
+except ImportError:
+    import json
+
+from flask import Flask
+import gevent
+from gevent import monkey
+from gevent.queue import Channel
+import os
+import logging
+import redis
+
+monkey.patch_all()
+
+application = Flask(__name__)
+application.config.from_object('tyron.default_settings')
+if 'TYRON_CONF' in os.environ:
+    application.config.from_envvar('TYRON_CONF')
+application.logger.setLevel(application.config['LOG_LEVEL'])
+
+subscriptions = defaultdict(Channel)
+
+class RedisSub(gevent.Greenlet):
+    """
+    subscribes to a redis pubsub channel and routes
+    messages to subscribers
+
+    messages have this format
+    {'channel': ..., 'data': ...}
+
+    """
+
+    def __init__(self, pubsub_channel, redis_hostname, redis_port, redis_db, redis_password=None):
+        gevent.Greenlet.__init__(self)
+        self.pubsub_channel = pubsub_channel
+        self.redis_hostname = redis_hostname
+        self.redis_port = redis_port
+        self.redis_db = redis_db
+        self.redis_password = redis_password
+
+    def get_redis_connection(self):
+        return redis.Redis(self.redis_hostname, self.redis_port, self.redis_db, self.redis_password)
+
+    def parse_message(self, message):
+        msg = json.loads(message)
+        return msg['channel'], msg['data']
+
+    def handle_message(self, message):
+        channel, data = self.parse_message(message)
+        gevent_channel = subscriptions[channel]
+        while gevent_channel.getters:
+            gevent_channel.put_nowait(data)
+
+    def subscribe(self):
+        connection = self.get_redis_connection()
+        pubsub = connection.pubsub()
+        pubsub.subscribe(self.pubsub_channel)
+        for message in pubsub.listen():
+            if message['type'] != 'message':
+                continue
+            self.handle_message(message['data'])
+
+    def _run(self):
+        self.subscribe()
+
+@application.route('/')
+def health():
+    return 'OK'
+
+@application.route('/<channel>/', methods=('GET', 'POST', 'OPTIONS'))
+def subscribe(channel):
+    timeout = application.config['LONGPOLLING_TIMEOUT']
+    try:
+        message = subscriptions[channel].get(timeout=timeout)
+    except Timeout:
+        message = application.config['TIMEOUT_RESPONSE_MESSAGE']
+    return message
+
+@application.before_first_request
+def start_subscribe_loop():
+    pubsub = RedisSub(
+        pubsub_channel=application.config['CHANNEL_NAME'],
+        redis_hostname=application.config['REDIS_HOSTNAME'],
+        redis_port=application.config['REDIS_PORT'],
+        redis_db=application.config['REDIS_DB']
+    )
+    gevent.spawn(pubsub.start)
+
+if __name__ == '__main__':
+    application.run()