From 6efa7f8d1d71ee6e2bfa46faea34d59e0f716154 Mon Sep 17 00:00:00 2001 From: Kyle Mahan Date: Tue, 24 Feb 2015 21:53:35 -0800 Subject: [PATCH] port websocket_server from Node to Tornado --- setup.py | 2 +- woodwind/static/feed.js | 26 ++++++++++++++ woodwind/static/websocket_client.js | 23 ------------- woodwind/tasks.py | 2 +- woodwind/templates/feed.jinja2 | 13 +++---- woodwind/views.py | 11 +++++- woodwind/websocket_server.py | 53 +++++++++++++++++++++++++++++ 7 files changed, 96 insertions(+), 34 deletions(-) delete mode 100644 woodwind/static/websocket_client.js create mode 100644 woodwind/websocket_server.py diff --git a/setup.py b/setup.py index 2e3b8bb..b727f20 100644 --- a/setup.py +++ b/setup.py @@ -12,4 +12,4 @@ setup(name='Woodwind', install_requires=[ 'Flask', 'Flask-Login', 'Flask-Micropub', 'Flask-SQLAlchemy', 'beautifulsoup4', 'bleach', 'celery', 'feedparser', 'html5lib', - 'mf2py', 'mf2util', 'redis', 'requests']) + 'mf2py', 'mf2util', 'redis', 'requests', 'tornado']) diff --git a/woodwind/static/feed.js b/woodwind/static/feed.js index 507d96d..52da4be 100644 --- a/woodwind/static/feed.js +++ b/woodwind/static/feed.js @@ -1,4 +1,5 @@ $(function(){ + function clickOlderLink(evt) { evt.preventDefault(); $.get(this.href, function(result) { @@ -66,5 +67,30 @@ $(function(){ }); } + + // topic will be user:id or feed:id + function webSocketSubscribe(topic) { + if ('WebSocket' in window) { + var ws = new WebSocket(window.location.origin + .replace(/https?:\/\//, 'ws://') + .replace(/(:\d+)?$/, ':8077')); + ws.onopen = function(event) { + // send the topic + console.log('subscribing to topic: ' + topic); + ws.send(topic); + }; + ws.onmessage = function(event) { + var data = JSON.parse(event.data); + data.entries.forEach(function(entryHtml) { + $('body main').prepend(entryHtml); + }); + attachListeners(); + }; + } + } + attachListeners(); + if (WS_TOPIC) { + webSocketSubscribe(WS_TOPIC); + } }); diff --git a/woodwind/static/websocket_client.js b/woodwind/static/websocket_client.js deleted file mode 100644 index 0ac2f65..0000000 --- a/woodwind/static/websocket_client.js +++ /dev/null @@ -1,23 +0,0 @@ - -// topic will be woodwind::user:id or woodwind::feed:id -function webSocketSubscribe(topic) { - if ('WebSocket' in window) { - - var ws = new WebSocket(window.location.origin - .replace(/https?:\/\//, 'ws://') - .replace(/(:\d+)?$/, ':8077')); - - ws.onopen = function(event) { - // send the topic - console.log('subscribing to topic: ' + topic); - ws.send(topic); - }; - - ws.onmessage = function(event) { - var data = JSON.parse(event.data); - data.entries.forEach(function(entryHtml) { - $('body main').prepend(entryHtml); - }); - }; - } -} diff --git a/woodwind/tasks.py b/woodwind/tasks.py index 180fc53..3b965ba 100644 --- a/woodwind/tasks.py +++ b/woodwind/tasks.py @@ -194,7 +194,7 @@ def notify_feed_updated(session, feed, entries): for e in entries ], }) - redis.publish('woodwind:user:{}'.format(user.id), message) + redis.publish('woodwind_notify', message) def is_content_equal(e1, e2): diff --git a/woodwind/templates/feed.jinja2 b/woodwind/templates/feed.jinja2 index c39cf8d..3a1c68e 100644 --- a/woodwind/templates/feed.jinja2 +++ b/woodwind/templates/feed.jinja2 @@ -1,7 +1,10 @@ {% extends "base.jinja2" %} {% block head %} - - + + {% if ws_topic %} + + {% endif %} + {% if current_user and current_user.settings and current_user.settings.get('reply-method') == 'indie-config' %} @@ -31,10 +34,4 @@ {% endif %} - {% if current_user.is_authenticated() %} - - {% endif %} - {% endblock body %} diff --git a/woodwind/views.py b/woodwind/views.py index c0085e5..d497b02 100644 --- a/woodwind/views.py +++ b/woodwind/views.py @@ -21,6 +21,8 @@ views = flask.Blueprint('views', __name__) def index(): page = int(flask.request.args.get('page', 1)) entries = [] + ws_topic = None + if flask_login.current_user.is_authenticated(): per_page = flask.current_app.config.get('PER_PAGE', 30) offset = (page - 1) * per_page @@ -32,12 +34,19 @@ def index(): if 'feed' in flask.request.args: feed_hex = flask.request.args.get('feed').encode() feed_url = binascii.unhexlify(feed_hex).decode('utf-8') + feed = Feed.query.filter_by(feed=feed_url).first() + if not feed: + flask.abort(404) entry_query = entry_query.filter(Feed.feed == feed_url) + ws_topic = 'feed:{}'.format(feed.id) + else: + ws_topic = 'user:{}'.format(flask_login.current_user.id) entries = entry_query.order_by(Entry.published.desc())\ .offset(offset).limit(per_page).all() - return flask.render_template('feed.jinja2', entries=entries, page=page) + return flask.render_template('feed.jinja2', entries=entries, page=page, + ws_topic=ws_topic) @views.route('/install') diff --git a/woodwind/websocket_server.py b/woodwind/websocket_server.py new file mode 100644 index 0000000..e670186 --- /dev/null +++ b/woodwind/websocket_server.py @@ -0,0 +1,53 @@ +import itertools +import json +import redis +import threading +import tornado.ioloop +import tornado.websocket + + +SUBSCRIBERS = {} + + +def redis_loop(): + r = redis.StrictRedis() + ps = r.pubsub() + ps.subscribe('woodwind_notify') + for message in ps.listen(): + if message['type'] == 'message': + msg_data = str(message.get('data'), 'utf-8') + msg_blob = json.loads(msg_data) + user_topic = 'user:{}'.format(msg_blob.get('user')) + feed_topic = 'feed:{}'.format(msg_blob.get('feed')) + for subscriber in itertools.chain( + SUBSCRIBERS.get(user_topic, []), + SUBSCRIBERS.get(feed_topic, [])): + tornado.ioloop.IOLoop.instance().add_callback( + lambda: subscriber.forward_message(msg_data)) + + +class WSHandler(tornado.websocket.WebSocketHandler): + def check_origin(self, origin): + return True + + def forward_message(self, message): + self.write_message(message) + + def on_message(self, topic): + self.topic = topic + SUBSCRIBERS.setdefault(topic, []).append(self) + + def on_close(self): + if hasattr(self, 'topic'): + SUBSCRIBERS.setdefault(self.topic, []).append(self) + + +application = tornado.web.Application([ + (r'/', WSHandler), +]) + + +if __name__ == '__main__': + threading.Thread(target=redis_loop).start() + application.listen(8077) + tornado.ioloop.IOLoop.instance().start()