{"action":"created","comment":{"url":"https://api.github.com/repos/fedora-infra/fmn.sse/pulls/comments/84385280","pull_request_review_id":5167860,"id":84385280,"diff_hunk":"@@ -0,0 +1,301 @@\n+from __future__ import absolute_import, unicode_literals\n+from collections import namedtuple\n+import json\n+import logging\n+\n+import fedmsg\n+import six\n+from pika import ConnectionParameters, exceptions as pika_exceptions\n+from pika.adapters import twisted_connection\n+from twisted.internet import reactor, defer, protocol, error\n+from twisted.web import server, resource\n+\n+\n+_log = logging.getLogger(__name__)\n+app_config = fedmsg.config.load_config()\n+\n+\n+# Create a few namedtuples to make the various tuples used throughout\n+# this module slightly clearer.\n+RabbitQueue = namedtuple(u'RabbitQueue', [u'queue', u'consumer_tag'])\n+SubscriptionEntry = namedtuple(u'SubscriptionEntry', [u'queue', u'requests'])\n+RequestKey = namedtuple(u'RequestKey', [u'exchange', u'queue'])\n+\n+\n+class SSEServer(resource.Resource):\n+    \"\"\"\n+    A Twisted Server-Sent Event server that consumes its messages from RabbitMQ.\n+\n+    There is only one endpoint for this server. Clients can request resources\n+    in the format ``/<RabbitMQ exchange>/<RabbitMQ queue>/``. An HTTP response\n+    with the SSE headers announcing it is an event stream are immediately\n+    returned. As messages are pushed to the RabbitMQ consumer, they are in turn\n+    pushed to all HTTP clients.\n+\n+    The response is never closed from the server's side. It continues to push\n+    events until the client disconnects.\n+    \"\"\"\n+    isLeaf = True\n+\n+    def __init__(self, *args, **kwargs):\n+        resource.Resource.__init__(self, *args, **kwargs)\n+        self.expire_ms = int(app_config.get('fmn.sse.pika.msg_expiration', 3600000))\n+        self.amqp_host = app_config.get('fmn.sse.pika.host', 'localhost')\n+        self.amqp_port = int(app_config.get('fmn.sse.pika.port', 5672))\n+        self.amqp_parameters = ConnectionParameters(self.amqp_host, self.amqp_port)\n+        cc = protocol.ClientCreator(\n+            reactor,\n+            twisted_connection.TwistedProtocolConnection,\n+            self.amqp_parameters,\n+        )\n+        self._deferred_connection = cc.connectTCP(self.amqp_host, self.amqp_port)\n+        self._deferred_connection.addCallback(lambda conn: conn.ready)\n+        self.connection = None\n+        self.channel = None\n+\n+        # Maps queues to open requests. The datastructure is in the format\n+        # {\n+        #   RequestKey: {'queue': RabbitQueue, 'requests': []},\n+        # }\n+        self.subscribers = {}\n+\n+    @defer.inlineCallbacks\n+    def queue(self, exchange=None, queue=None, routing_key=None,\n+              queue_arguments=None):\n+        \"\"\"\n+        Construct and retrieve a queue from RabbitMQ.\n+\n+        This will declare the exchange and queue in RabbitMQ if they don't\n+        already exist.\n+\n+        The queue returned is a subclass of twisted.internet.defer.DeferredQueue.\n+        Calling ``get`` on the queue returns a ``Deferred`` that fires with\n+        (channel, method, properties, body) where the types are\n+        (pika.Channel, pika.spec.Basic.Deliver, pika.spec.BasicProperties, bytes).\n+\n+        Afterwards you can do channel.basic_ack(delivery_tag=method.delivery_tag)\n+        to acknowledge message reciept.\n+\n+        :param exchange:        The exchange to declare and bind the queue to.\n+        :type  exchange:        unicode str\n+        :param queue:           The queue to declare, bind to the exchange,\n+                                and consume from.\n+        :type  queue:           unicode str\n+        :param routing_key:     The queue routing key.\n+        :type  routing_key:     unicode str\n+        :param queue_arguments: custom arguments for the queue's construction\n+        :type  queue_agrements: dict\n+\n+        :return: A Deferred that fires with a tuple of the queue to consume\n+                 messages from and a consumer tag.\n+        :rtype: Deferred of (pika.ClosableDeferredQueue, unicode str)\n+        \"\"\"\n+        # TODO Handle errors appropriately\n+        if not self.connection:\n+            self.connection = yield self._deferred_connection\n+        if not self.channel:\n+            self.channel = yield self.connection.channel()\n+\n+        # TODO Consider using ``passive=True`` on ``exchange_declare`` and\n+        # ``queue_declare``. This causes an error if the queue or exchange don't\n+        # already exist, which we could then turn into a HTTP 404. Allowing\n+        # requests to cause exchanges or queues to be created seems like an easy\n+        # DoS attack.\n+        yield self.channel.exchange_declare(exchange=exchange)\n+        yield self.channel.queue_declare(\n+            queue=queue,\n+            durable=True,\n+            auto_delete=False,\n+            arguments={'x-message-ttl': self.expire_ms}\n+        )\n+        # TODO investigate why this routing_key is used","path":"fmn/sse/server.py","position":111,"original_position":111,"commit_id":"f2daa5fe6e3c0ef3e7932b5217758488412753fe","original_commit_id":"f2daa5fe6e3c0ef3e7932b5217758488412753fe","user":{"login":"skrzepto","id":3166193,"avatar_url":"https://avatars.githubusercontent.com/u/3166193?v=3","gravatar_id":"","url":"https://api.github.com/users/skrzepto","html_url":"https://github.com/skrzepto","followers_url":"https://api.github.com/users/skrzepto/followers","following_url":"https://api.github.com/users/skrzepto/following{/other_user}","gists_url":"https://api.github.com/users/skrzepto/gists{/gist_id}","starred_url":"https://api.github.com/users/skrzepto/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/skrzepto/subscriptions","organizations_url":"https://api.github.com/users/skrzepto/orgs","repos_url":"https://api.github.com/users/skrzepto/repos","events_url":"https://api.github.com/users/skrzepto/events{/privacy}","received_events_url":"https://api.github.com/users/skrzepto/received_events","type":"User","site_admin":false},"body":"to prevent collision between a user and a group name having the same identifier. ","created_at":"2016-10-20T22:03:38Z","updated_at":"2016-10-20T22:03:38Z","html_url":"https://github.com/fedora-infra/fmn.sse/pull/52#discussion_r84385280","pull_request_url":"https://api.github.com/repos/fedora-infra/fmn.sse/pulls/52","_links":{"self":{"href":"https://api.github.com/repos/fedora-infra/fmn.sse/pulls/comments/84385280"},"html":{"href":"https://github.com/fedora-infra/fmn.sse/pull/52#discussion_r84385280"},"pull_request":{"href":"https://api.github.com/repos/fedora-infra/fmn.sse/pulls/52"}}},"pull_request":{"url":"https://api.github.com/repos/fedora-infra/fmn.sse/pulls/52","id":90084100,"html_url":"https://github.com/fedora-infra/fmn.sse/pull/52","diff_url":"https://github.com/fedora-infra/fmn.sse/pull/52.diff","patch_url":"https://github.com/fedora-infra/fmn.sse/pull/52.patch","issue_url":"https://api.github.com/repos/fedora-infra/fmn.sse/issues/52","number":52,"state":"open","locked":false,"title":"Convert the SSE server to a Twisted pika connection","user":{"login":"jeremycline","id":1977525,"avatar_url":"https://avatars.githubusercontent.com/u/1977525?v=3","gravatar_id":"","url":"https://api.github.com/users/jeremycline","html_url":"https://github.com/jeremycline","followers_url":"https://api.github.com/users/jeremycline/followers","following_url":"https://api.github.com/users/jeremycline/following{/other_user}","gists_url":"https://api.github.com/users/jeremycline/gists{/gist_id}","starred_url":"https://api.github.com/users/jeremycline/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jeremycline/subscriptions","organizations_url":"https://api.github.com/users/jeremycline/orgs","repos_url":"https://api.github.com/users/jeremycline/repos","events_url":"https://api.github.com/users/jeremycline/events{/privacy}","received_events_url":"https://api.github.com/users/jeremycline/received_events","type":"User","site_admin":false},"body":"Prior to this commit, a blocking connection to RabbitMQ was used. This\r\nis problematic in Twisted because it blocks the reactor rather than take\r\nadvantage of Twisted's reactor design.\r\n    \r\nThis should behave almost exactly like the previous connection, except\r\nthat now messages are pushed by RabbitMQ to the SSE server and in turn\r\nimmediately pushed out to clients. It also works with a single TCP\r\nconnection rather than using one connection per queue.\r\n    \r\nFixes #48 and #50\r\n    \r\nSigned-off-by: Jeremy Cline <jeremy@jcline.org>\r\n\r\n\r\nA few notes for reviewers:\r\n* I'm really sorry for the huge diff. I didn't want it to be so large, but I couldn't just drop the new non-blocking connection without touching all the things that used it.\r\n\r\n* I've based this branch off a branch that hasn't been merged quite yet so right now it's pulling in a few additional commits. The last one is the only one of interest here.\r\n\r\n* This PR is currently a work in progress, but I wanted to give people an opportunity to look at it and comment early on. This should be functional, if a little unpolished. There are quite a few ``# TODO`` comments, and they'll all go away. I'll make a note when I think it's ready for a final review.\r\n\r\n* I tried to translate 1-to-1 what was there previously with what's here now, so it should behave the same way in terms of exchanges, queues, and their configurations.\r\n\r\nLet me know what you all think!\r\n","created_at":"2016-10-19T20:27:31Z","updated_at":"2016-10-20T22:03:38Z","closed_at":null,"merged_at":null,"merge_commit_sha":"a266e48c69d1de3608a9ce9c48772522adee76c0","assignee":null,"assignees":[],"milestone":null,"commits_url":"https://api.github.com/repos/fedora-infra/fmn.sse/pulls/52/commits","review_comments_url":"https://api.github.com/repos/fedora-infra/fmn.sse/pulls/52/comments","review_comment_url":"https://api.github.com/repos/fedora-infra/fmn.sse/pulls/comments{/number}","comments_url":"https://api.github.com/repos/fedora-infra/fmn.sse/issues/52/comments","statuses_url":"https://api.github.com/repos/fedora-infra/fmn.sse/statuses/f2daa5fe6e3c0ef3e7932b5217758488412753fe","head":{"label":"jeremycline:twisted-connection","ref":"twisted-connection","sha":"f2daa5fe6e3c0ef3e7932b5217758488412753fe","user":{"login":"jeremycline","id":1977525,"avatar_url":"https://avatars.githubusercontent.com/u/1977525?v=3","gravatar_id":"","url":"https://api.github.com/users/jeremycline","html_url":"https://github.com/jeremycline","followers_url":"https://api.github.com/users/jeremycline/followers","following_url":"https://api.github.com/users/jeremycline/following{/other_user}","gists_url":"https://api.github.com/users/jeremycline/gists{/gist_id}","starred_url":"https://api.github.com/users/jeremycline/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jeremycline/subscriptions","organizations_url":"https://api.github.com/users/jeremycline/orgs","repos_url":"https://api.github.com/users/jeremycline/repos","events_url":"https://api.github.com/users/jeremycline/events{/privacy}","received_events_url":"https://api.github.com/users/jeremycline/received_events","type":"User","site_admin":false},"repo":{"id":70199129,"name":"fmn.sse","full_name":"jeremycline/fmn.sse","owner":{"login":"jeremycline","id":1977525,"avatar_url":"https://avatars.githubusercontent.com/u/1977525?v=3","gravatar_id":"","url":"https://api.github.com/users/jeremycline","html_url":"https://github.com/jeremycline","followers_url":"https://api.github.com/users/jeremycline/followers","following_url":"https://api.github.com/users/jeremycline/following{/other_user}","gists_url":"https://api.github.com/users/jeremycline/gists{/gist_id}","starred_url":"https://api.github.com/users/jeremycline/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jeremycline/subscriptions","organizations_url":"https://api.github.com/users/jeremycline/orgs","repos_url":"https://api.github.com/users/jeremycline/repos","events_url":"https://api.github.com/users/jeremycline/events{/privacy}","received_events_url":"https://api.github.com/users/jeremycline/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/jeremycline/fmn.sse","description":"SSE server for realtime feed data","fork":true,"url":"https://api.github.com/repos/jeremycline/fmn.sse","forks_url":"https://api.github.com/repos/jeremycline/fmn.sse/forks","keys_url":"https://api.github.com/repos/jeremycline/fmn.sse/keys{/key_id}","collaborators_url":"https://api.github.com/repos/jeremycline/fmn.sse/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/jeremycline/fmn.sse/teams","hooks_url":"https://api.github.com/repos/jeremycline/fmn.sse/hooks","issue_events_url":"https://api.github.com/repos/jeremycline/fmn.sse/issues/events{/number}","events_url":"https://api.github.com/repos/jeremycline/fmn.sse/events","assignees_url":"https://api.github.com/repos/jeremycline/fmn.sse/assignees{/user}","branches_url":"https://api.github.com/repos/jeremycline/fmn.sse/branches{/branch}","tags_url":"https://api.github.com/repos/jeremycline/fmn.sse/tags","blobs_url":"https://api.github.com/repos/jeremycline/fmn.sse/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/jeremycline/fmn.sse/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/jeremycline/fmn.sse/git/refs{/sha}","trees_url":"https://api.github.com/repos/jeremycline/fmn.sse/git/trees{/sha}","statuses_url":"https://api.github.com/repos/jeremycline/fmn.sse/statuses/{sha}","languages_url":"https://api.github.com/repos/jeremycline/fmn.sse/languages","stargazers_url":"https://api.github.com/repos/jeremycline/fmn.sse/stargazers","contributors_url":"https://api.github.com/repos/jeremycline/fmn.sse/contributors","subscribers_url":"https://api.github.com/repos/jeremycline/fmn.sse/subscribers","subscription_url":"https://api.github.com/repos/jeremycline/fmn.sse/subscription","commits_url":"https://api.github.com/repos/jeremycline/fmn.sse/commits{/sha}","git_commits_url":"https://api.github.com/repos/jeremycline/fmn.sse/git/commits{/sha}","comments_url":"https://api.github.com/repos/jeremycline/fmn.sse/comments{/number}","issue_comment_url":"https://api.github.com/repos/jeremycline/fmn.sse/issues/comments{/number}","contents_url":"https://api.github.com/repos/jeremycline/fmn.sse/contents/{+path}","compare_url":"https://api.github.com/repos/jeremycline/fmn.sse/compare/{base}...{head}","merges_url":"https://api.github.com/repos/jeremycline/fmn.sse/merges","archive_url":"https://api.github.com/repos/jeremycline/fmn.sse/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/jeremycline/fmn.sse/downloads","issues_url":"https://api.github.com/repos/jeremycline/fmn.sse/issues{/number}","pulls_url":"https://api.github.com/repos/jeremycline/fmn.sse/pulls{/number}","milestones_url":"https://api.github.com/repos/jeremycline/fmn.sse/milestones{/number}","notifications_url":"https://api.github.com/repos/jeremycline/fmn.sse/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/jeremycline/fmn.sse/labels{/name}","releases_url":"https://api.github.com/repos/jeremycline/fmn.sse/releases{/id}","deployments_url":"https://api.github.com/repos/jeremycline/fmn.sse/deployments","created_at":"2016-10-06T22:48:33Z","updated_at":"2016-10-06T22:48:34Z","pushed_at":"2016-10-20T19:16:59Z","git_url":"git://github.com/jeremycline/fmn.sse.git","ssh_url":"git@github.com:jeremycline/fmn.sse.git","clone_url":"https://github.com/jeremycline/fmn.sse.git","svn_url":"https://github.com/jeremycline/fmn.sse","homepage":null,"size":129,"stargazers_count":0,"watchers_count":0,"language":"Python","has_issues":false,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"fedora-infra:develop","ref":"develop","sha":"48945437cda09c7852b315b79f1f32ac3a3d7402","user":{"login":"fedora-infra","id":3316637,"avatar_url":"https://avatars.githubusercontent.com/u/3316637?v=3","gravatar_id":"","url":"https://api.github.com/users/fedora-infra","html_url":"https://github.com/fedora-infra","followers_url":"https://api.github.com/users/fedora-infra/followers","following_url":"https://api.github.com/users/fedora-infra/following{/other_user}","gists_url":"https://api.github.com/users/fedora-infra/gists{/gist_id}","starred_url":"https://api.github.com/users/fedora-infra/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/fedora-infra/subscriptions","organizations_url":"https://api.github.com/users/fedora-infra/orgs","repos_url":"https://api.github.com/users/fedora-infra/repos","events_url":"https://api.github.com/users/fedora-infra/events{/privacy}","received_events_url":"https://api.github.com/users/fedora-infra/received_events","type":"Organization","site_admin":false},"repo":{"id":62312816,"name":"fmn.sse","full_name":"fedora-infra/fmn.sse","owner":{"login":"fedora-infra","id":3316637,"avatar_url":"https://avatars.githubusercontent.com/u/3316637?v=3","gravatar_id":"","url":"https://api.github.com/users/fedora-infra","html_url":"https://github.com/fedora-infra","followers_url":"https://api.github.com/users/fedora-infra/followers","following_url":"https://api.github.com/users/fedora-infra/following{/other_user}","gists_url":"https://api.github.com/users/fedora-infra/gists{/gist_id}","starred_url":"https://api.github.com/users/fedora-infra/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/fedora-infra/subscriptions","organizations_url":"https://api.github.com/users/fedora-infra/orgs","repos_url":"https://api.github.com/users/fedora-infra/repos","events_url":"https://api.github.com/users/fedora-infra/events{/privacy}","received_events_url":"https://api.github.com/users/fedora-infra/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/fedora-infra/fmn.sse","description":"SSE server for realtime feed data","fork":false,"url":"https://api.github.com/repos/fedora-infra/fmn.sse","forks_url":"https://api.github.com/repos/fedora-infra/fmn.sse/forks","keys_url":"https://api.github.com/repos/fedora-infra/fmn.sse/keys{/key_id}","collaborators_url":"https://api.github.com/repos/fedora-infra/fmn.sse/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/fedora-infra/fmn.sse/teams","hooks_url":"https://api.github.com/repos/fedora-infra/fmn.sse/hooks","issue_events_url":"https://api.github.com/repos/fedora-infra/fmn.sse/issues/events{/number}","events_url":"https://api.github.com/repos/fedora-infra/fmn.sse/events","assignees_url":"https://api.github.com/repos/fedora-infra/fmn.sse/assignees{/user}","branches_url":"https://api.github.com/repos/fedora-infra/fmn.sse/branches{/branch}","tags_url":"https://api.github.com/repos/fedora-infra/fmn.sse/tags","blobs_url":"https://api.github.com/repos/fedora-infra/fmn.sse/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/fedora-infra/fmn.sse/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/fedora-infra/fmn.sse/git/refs{/sha}","trees_url":"https://api.github.com/repos/fedora-infra/fmn.sse/git/trees{/sha}","statuses_url":"https://api.github.com/repos/fedora-infra/fmn.sse/statuses/{sha}","languages_url":"https://api.github.com/repos/fedora-infra/fmn.sse/languages","stargazers_url":"https://api.github.com/repos/fedora-infra/fmn.sse/stargazers","contributors_url":"https://api.github.com/repos/fedora-infra/fmn.sse/contributors","subscribers_url":"https://api.github.com/repos/fedora-infra/fmn.sse/subscribers","subscription_url":"https://api.github.com/repos/fedora-infra/fmn.sse/subscription","commits_url":"https://api.github.com/repos/fedora-infra/fmn.sse/commits{/sha}","git_commits_url":"https://api.github.com/repos/fedora-infra/fmn.sse/git/commits{/sha}","comments_url":"https://api.github.com/repos/fedora-infra/fmn.sse/comments{/number}","issue_comment_url":"https://api.github.com/repos/fedora-infra/fmn.sse/issues/comments{/number}","contents_url":"https://api.github.com/repos/fedora-infra/fmn.sse/contents/{+path}","compare_url":"https://api.github.com/repos/fedora-infra/fmn.sse/compare/{base}...{head}","merges_url":"https://api.github.com/repos/fedora-infra/fmn.sse/merges","archive_url":"https://api.github.com/repos/fedora-infra/fmn.sse/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/fedora-infra/fmn.sse/downloads","issues_url":"https://api.github.com/repos/fedora-infra/fmn.sse/issues{/number}","pulls_url":"https://api.github.com/repos/fedora-infra/fmn.sse/pulls{/number}","milestones_url":"https://api.github.com/repos/fedora-infra/fmn.sse/milestones{/number}","notifications_url":"https://api.github.com/repos/fedora-infra/fmn.sse/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/fedora-infra/fmn.sse/labels{/name}","releases_url":"https://api.github.com/repos/fedora-infra/fmn.sse/releases{/id}","deployments_url":"https://api.github.com/repos/fedora-infra/fmn.sse/deployments","created_at":"2016-06-30T13:29:12Z","updated_at":"2016-10-10T15:53:57Z","pushed_at":"2016-10-20T19:17:00Z","git_url":"git://github.com/fedora-infra/fmn.sse.git","ssh_url":"git@github.com:fedora-infra/fmn.sse.git","clone_url":"https://github.com/fedora-infra/fmn.sse.git","svn_url":"https://github.com/fedora-infra/fmn.sse","homepage":null,"size":133,"stargazers_count":0,"watchers_count":0,"language":"Python","has_issues":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":2,"mirror_url":null,"open_issues_count":15,"forks":2,"open_issues":15,"watchers":0,"default_branch":"develop"}},"_links":{"self":{"href":"https://api.github.com/repos/fedora-infra/fmn.sse/pulls/52"},"html":{"href":"https://github.com/fedora-infra/fmn.sse/pull/52"},"issue":{"href":"https://api.github.com/repos/fedora-infra/fmn.sse/issues/52"},"comments":{"href":"https://api.github.com/repos/fedora-infra/fmn.sse/issues/52/comments"},"review_comments":{"href":"https://api.github.com/repos/fedora-infra/fmn.sse/pulls/52/comments"},"review_comment":{"href":"https://api.github.com/repos/fedora-infra/fmn.sse/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/fedora-infra/fmn.sse/pulls/52/commits"},"statuses":{"href":"https://api.github.com/repos/fedora-infra/fmn.sse/statuses/f2daa5fe6e3c0ef3e7932b5217758488412753fe"}}}}
{"actor":{"display_login":"skrzepto"}}