{"action":"created","comment":{"url":"https://api.github.com/repos/leapcode/leap_pycommon/pulls/comments/26310092","id":26310092,"diff_hunk":"@@ -55,275 +54,269 @@ class CallbackAlreadyRegistered(Exception):\n     pass\n \n \n-def ensure_client_daemon():\n+class EventsIOLoop(ioloop.ZMQIOLoop):\n     \"\"\"\n-    Ensure the client daemon is running and listening for incoming\n-    messages.\n-\n-    :return: the daemon instance\n-    :rtype: EventsClientDaemon\n+    An extension of zmq's ioloop that can wait until there are no callbacks\n+    in the queue before stopping.\n     \"\"\"\n-    import time\n-    daemon = EventsClientDaemon.ensure(0)\n-    logger.debug('ensure client daemon')\n \n-    # Because we use a random port we want to wait until a port is assigned to\n-    # local client daemon.\n+    def stop(self, wait=False):\n+        \"\"\"\n+        Stop the I/O loop.\n \n-    while not (EventsClientDaemon.get_instance() and\n-               EventsClientDaemon.get_instance().get_port()):\n-        time.sleep(0.1)\n-    return daemon\n+        :param wait: Whether we should wait for callbacks in queue to finish\n+                     before stopping.\n+        :type wait: bool\n+        \"\"\"\n+        if wait:\n+            # prevent new callbacks from being added\n+            with self._callback_lock:\n+                self._closing = True\n+            # wait until all callbacks have been executed\n+            while self._callbacks:\n+                time.sleep(0.1)\n+        ioloop.ZMQIOLoop.stop(self)\n \n \n-def register(signal, callback, uid=None, replace=False, reqcbk=None,\n-             timeout=1000):\n-    \"\"\"\n-    Registers a callback to be called when a specific signal event is\n-    received.\n-\n-    Will timeout after timeout ms if response has not been received. The\n-    timeout arg is only used for asynch requests. If a reqcbk callback has\n-    been supplied the timeout arg is not used. The response value will be\n-    returned for a synch request but nothing will be returned for an asynch\n-    request.\n-\n-    :param signal: the signal that causes the callback to be launched\n-    :type signal: int (see the `events.proto` file)\n-    :param callback: the callback to be called when the signal is received\n-    :type callback: function(leap.common.events.events_pb2.SignalRequest)\n-    :param uid: a unique id for the callback\n-    :type uid: int\n-    :param replace: should an existent callback with same uid be replaced?\n-    :type replace: bool\n-    :param reqcbk: a callback to be called when a response from server is\n-                   received\n-    :type reqcbk: function(proto.RegisterRequest, proto.EventResponse)\n-    :param timeout: the timeout for synch calls\n-    :type timeout: int\n-\n-    Might raise a CallbackAlreadyRegistered exception if there's already a\n-    callback identified by the given uid and replace is False.\n-\n-    :return: the response from server for synch calls or nothing for asynch\n-             calls.\n-    :rtype: leap.common.events.events_pb2.EventsResponse or None\n-    \"\"\"\n-    ensure_client_daemon()  # so we can receive registered signals\n-    # register callback locally\n-    if signal not in registered_callbacks:\n-        registered_callbacks[signal] = []\n-    cbklist = registered_callbacks[signal]\n+class EventsClient(threading.Thread):\n \n-    # TODO should check that the callback has the right\n-    # number of arguments.\n+    _instance = None\n+    _instance_lock = threading.Lock()\n \n-    if uid and filter(lambda (x, y): x == uid, cbklist):\n-        if not replace:\n-            raise CallbackAlreadyRegistered()\n-        else:\n-            registered_callbacks[signal] = filter(lambda(x, y): x != uid,\n-                                                  cbklist)\n-    registered_callbacks[signal].append((uid, callback))\n-    # register callback on server\n-    request = proto.RegisterRequest()\n-    request.event = signal\n-    request.port = EventsClientDaemon.get_instance().get_port()\n-    request.mac_method = mac_auth.MacMethod.MAC_NONE\n-    request.mac = \"\"\n-    service = RpcService(proto.EventsServerService_Stub,\n-                         server.SERVER_PORT, 'localhost')\n-    logger.debug(\n-        \"Sending registration request to server on port %s: %s\",\n-        server.SERVER_PORT,\n-        str(request)[:40])\n-    return service.register(request, callback=reqcbk, timeout=timeout)\n-\n-\n-def unregister(signal, uid=None, reqcbk=None, timeout=1000):\n-    \"\"\"\n-    Unregister a callback.\n-\n-    If C{uid} is specified, unregisters only the callback identified by that\n-    unique id. Otherwise, unregisters all callbacks\n-\n-    :param signal: the signal that causes the callback to be launched\n-    :type signal: int (see the `events.proto` file)\n-    :param uid: a unique id for the callback\n-    :type uid: int\n-    :param reqcbk: a callback to be called when a response from server is\n-                   received\n-    :type reqcbk: function(proto.UnregisterRequest, proto.EventResponse)\n-    :param timeout: the timeout for synch calls\n-    :type timeout: int\n-\n-    :return: the response from server for synch calls or nothing for asynch\n-             calls or None if no callback is registered for that signal or\n-             uid.\n-    :rtype: leap.common.events.events_pb2.EventsResponse or None\n-    \"\"\"\n-    if signal not in registered_callbacks or not registered_callbacks[signal]:\n-        logger.warning(\"No callback registered for signal %d.\" % signal)\n-        return None\n-    # unregister callback locally\n-    cbklist = registered_callbacks[signal]\n-    if uid is not None:\n-        if filter(lambda (cbkuid, _): cbkuid == uid, cbklist) == []:\n-            logger.warning(\"No callback registered for uid %d.\" % st)\n-            return None\n-        registered_callbacks[signal] = filter(lambda(x, y): x != uid, cbklist)\n-    else:\n-        # exclude all callbacks for given signal\n-        registered_callbacks[signal] = []\n-    # unregister port in server if there are no more callbacks for this signal\n-    if not registered_callbacks[signal]:\n-        request = proto.UnregisterRequest()\n-        request.event = signal\n-        request.port = EventsClientDaemon.get_instance().get_port()\n-        request.mac_method = mac_auth.MacMethod.MAC_NONE\n-        request.mac = \"\"\n-        service = RpcService(proto.EventsServerService_Stub,\n-                             server.SERVER_PORT, 'localhost')\n-        logger.info(\n-            \"Sending unregistration request to server on port %s: %s\",\n-            server.SERVER_PORT,\n-            str(request)[:40])\n-        return service.unregister(request, callback=reqcbk, timeout=timeout)\n-\n-\n-def signal(signal, content=\"\", mac_method=\"\", mac=\"\", reqcbk=None,\n-           timeout=1000):\n-    \"\"\"\n-    Send `signal` event to events server.\n+    def __init__(self):\n+        \"\"\"\n+        Initialize the events client.\n+        \"\"\"\n+        threading.Thread.__init__(self)\n+        self._callbacks = collections.defaultdict(list)\n+        self._lock = threading.Lock()\n+        self._initialized = threading.Event()\n \n-    Will timeout after timeout ms if response has not been received. The\n-    timeout arg is only used for asynch requests.  If a reqcbk callback has\n-    been supplied the timeout arg is not used. The response value will be\n-    returned for a synch request but nothing will be returned for an asynch\n-    request.\n+    @classmethod\n+    def instance(self):","path":"src/leap/common/events/client.py","position":237,"original_position":237,"commit_id":"801ce9890c951f6ffda011b38cec3b2f67d398b9","original_commit_id":"801ce9890c951f6ffda011b38cec3b2f67d398b9","user":{"login":"andrejb","id":79672,"avatar_url":"https://avatars.githubusercontent.com/u/79672?v=3","gravatar_id":"","url":"https://api.github.com/users/andrejb","html_url":"https://github.com/andrejb","followers_url":"https://api.github.com/users/andrejb/followers","following_url":"https://api.github.com/users/andrejb/following{/other_user}","gists_url":"https://api.github.com/users/andrejb/gists{/gist_id}","starred_url":"https://api.github.com/users/andrejb/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/andrejb/subscriptions","organizations_url":"https://api.github.com/users/andrejb/orgs","repos_url":"https://api.github.com/users/andrejb/repos","events_url":"https://api.github.com/users/andrejb/events{/privacy}","received_events_url":"https://api.github.com/users/andrejb/received_events","type":"User","site_admin":false},"body":"You're correct. :-)","created_at":"2015-03-12T15:12:54Z","updated_at":"2015-03-12T15:12:54Z","html_url":"https://github.com/leapcode/leap_pycommon/pull/95#discussion_r26310092","pull_request_url":"https://api.github.com/repos/leapcode/leap_pycommon/pulls/95","_links":{"self":{"href":"https://api.github.com/repos/leapcode/leap_pycommon/pulls/comments/26310092"},"html":{"href":"https://github.com/leapcode/leap_pycommon/pull/95#discussion_r26310092"},"pull_request":{"href":"https://api.github.com/repos/leapcode/leap_pycommon/pulls/95"}}},"pull_request":{"url":"https://api.github.com/repos/leapcode/leap_pycommon/pulls/95","id":30859681,"html_url":"https://github.com/leapcode/leap_pycommon/pull/95","diff_url":"https://github.com/leapcode/leap_pycommon/pull/95.diff","patch_url":"https://github.com/leapcode/leap_pycommon/pull/95.patch","issue_url":"https://api.github.com/repos/leapcode/leap_pycommon/issues/95","number":95,"state":"open","locked":false,"title":"[feature] Refactor leap.common.events to use ZMQ.","user":{"login":"andrejb","id":79672,"avatar_url":"https://avatars.githubusercontent.com/u/79672?v=3","gravatar_id":"","url":"https://api.github.com/users/andrejb","html_url":"https://github.com/andrejb","followers_url":"https://api.github.com/users/andrejb/followers","following_url":"https://api.github.com/users/andrejb/following{/other_user}","gists_url":"https://api.github.com/users/andrejb/gists{/gist_id}","starred_url":"https://api.github.com/users/andrejb/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/andrejb/subscriptions","organizations_url":"https://api.github.com/users/andrejb/orgs","repos_url":"https://api.github.com/users/andrejb/repos","events_url":"https://api.github.com/users/andrejb/events{/privacy}","received_events_url":"https://api.github.com/users/andrejb/received_events","type":"User","site_admin":false},"body":"Closes: #6359","created_at":"2015-03-10T15:17:18Z","updated_at":"2015-03-12T15:12:54Z","closed_at":null,"merged_at":null,"merge_commit_sha":"a838db9d8beafe20f9afe31efcb5718d33f67b91","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/leapcode/leap_pycommon/pulls/95/commits","review_comments_url":"https://api.github.com/repos/leapcode/leap_pycommon/pulls/95/comments","review_comment_url":"https://api.github.com/repos/leapcode/leap_pycommon/pulls/comments{/number}","comments_url":"https://api.github.com/repos/leapcode/leap_pycommon/issues/95/comments","statuses_url":"https://api.github.com/repos/leapcode/leap_pycommon/statuses/801ce9890c951f6ffda011b38cec3b2f67d398b9","head":{"label":"andrejb:feature/6359_modify-events-to-use-zmq","ref":"feature/6359_modify-events-to-use-zmq","sha":"801ce9890c951f6ffda011b38cec3b2f67d398b9","user":{"login":"andrejb","id":79672,"avatar_url":"https://avatars.githubusercontent.com/u/79672?v=3","gravatar_id":"","url":"https://api.github.com/users/andrejb","html_url":"https://github.com/andrejb","followers_url":"https://api.github.com/users/andrejb/followers","following_url":"https://api.github.com/users/andrejb/following{/other_user}","gists_url":"https://api.github.com/users/andrejb/gists{/gist_id}","starred_url":"https://api.github.com/users/andrejb/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/andrejb/subscriptions","organizations_url":"https://api.github.com/users/andrejb/orgs","repos_url":"https://api.github.com/users/andrejb/repos","events_url":"https://api.github.com/users/andrejb/events{/privacy}","received_events_url":"https://api.github.com/users/andrejb/received_events","type":"User","site_admin":false},"repo":{"id":8959712,"name":"leap_pycommon","full_name":"andrejb/leap_pycommon","owner":{"login":"andrejb","id":79672,"avatar_url":"https://avatars.githubusercontent.com/u/79672?v=3","gravatar_id":"","url":"https://api.github.com/users/andrejb","html_url":"https://github.com/andrejb","followers_url":"https://api.github.com/users/andrejb/followers","following_url":"https://api.github.com/users/andrejb/following{/other_user}","gists_url":"https://api.github.com/users/andrejb/gists{/gist_id}","starred_url":"https://api.github.com/users/andrejb/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/andrejb/subscriptions","organizations_url":"https://api.github.com/users/andrejb/orgs","repos_url":"https://api.github.com/users/andrejb/repos","events_url":"https://api.github.com/users/andrejb/events{/privacy}","received_events_url":"https://api.github.com/users/andrejb/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/andrejb/leap_pycommon","description":"","fork":true,"url":"https://api.github.com/repos/andrejb/leap_pycommon","forks_url":"https://api.github.com/repos/andrejb/leap_pycommon/forks","keys_url":"https://api.github.com/repos/andrejb/leap_pycommon/keys{/key_id}","collaborators_url":"https://api.github.com/repos/andrejb/leap_pycommon/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/andrejb/leap_pycommon/teams","hooks_url":"https://api.github.com/repos/andrejb/leap_pycommon/hooks","issue_events_url":"https://api.github.com/repos/andrejb/leap_pycommon/issues/events{/number}","events_url":"https://api.github.com/repos/andrejb/leap_pycommon/events","assignees_url":"https://api.github.com/repos/andrejb/leap_pycommon/assignees{/user}","branches_url":"https://api.github.com/repos/andrejb/leap_pycommon/branches{/branch}","tags_url":"https://api.github.com/repos/andrejb/leap_pycommon/tags","blobs_url":"https://api.github.com/repos/andrejb/leap_pycommon/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/andrejb/leap_pycommon/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/andrejb/leap_pycommon/git/refs{/sha}","trees_url":"https://api.github.com/repos/andrejb/leap_pycommon/git/trees{/sha}","statuses_url":"https://api.github.com/repos/andrejb/leap_pycommon/statuses/{sha}","languages_url":"https://api.github.com/repos/andrejb/leap_pycommon/languages","stargazers_url":"https://api.github.com/repos/andrejb/leap_pycommon/stargazers","contributors_url":"https://api.github.com/repos/andrejb/leap_pycommon/contributors","subscribers_url":"https://api.github.com/repos/andrejb/leap_pycommon/subscribers","subscription_url":"https://api.github.com/repos/andrejb/leap_pycommon/subscription","commits_url":"https://api.github.com/repos/andrejb/leap_pycommon/commits{/sha}","git_commits_url":"https://api.github.com/repos/andrejb/leap_pycommon/git/commits{/sha}","comments_url":"https://api.github.com/repos/andrejb/leap_pycommon/comments{/number}","issue_comment_url":"https://api.github.com/repos/andrejb/leap_pycommon/issues/comments{/number}","contents_url":"https://api.github.com/repos/andrejb/leap_pycommon/contents/{+path}","compare_url":"https://api.github.com/repos/andrejb/leap_pycommon/compare/{base}...{head}","merges_url":"https://api.github.com/repos/andrejb/leap_pycommon/merges","archive_url":"https://api.github.com/repos/andrejb/leap_pycommon/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/andrejb/leap_pycommon/downloads","issues_url":"https://api.github.com/repos/andrejb/leap_pycommon/issues{/number}","pulls_url":"https://api.github.com/repos/andrejb/leap_pycommon/pulls{/number}","milestones_url":"https://api.github.com/repos/andrejb/leap_pycommon/milestones{/number}","notifications_url":"https://api.github.com/repos/andrejb/leap_pycommon/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/andrejb/leap_pycommon/labels{/name}","releases_url":"https://api.github.com/repos/andrejb/leap_pycommon/releases{/id}","created_at":"2013-03-22T19:54:17Z","updated_at":"2014-05-29T18:01:51Z","pushed_at":"2015-03-10T15:16:54Z","git_url":"git://github.com/andrejb/leap_pycommon.git","ssh_url":"git@github.com:andrejb/leap_pycommon.git","clone_url":"https://github.com/andrejb/leap_pycommon.git","svn_url":"https://github.com/andrejb/leap_pycommon","homepage":null,"size":788,"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":"develop"}},"base":{"label":"leapcode:develop","ref":"develop","sha":"60bcc511b5d715b4921645cad6c82adfacf7ec24","user":{"login":"leapcode","id":1700464,"avatar_url":"https://avatars.githubusercontent.com/u/1700464?v=3","gravatar_id":"","url":"https://api.github.com/users/leapcode","html_url":"https://github.com/leapcode","followers_url":"https://api.github.com/users/leapcode/followers","following_url":"https://api.github.com/users/leapcode/following{/other_user}","gists_url":"https://api.github.com/users/leapcode/gists{/gist_id}","starred_url":"https://api.github.com/users/leapcode/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/leapcode/subscriptions","organizations_url":"https://api.github.com/users/leapcode/orgs","repos_url":"https://api.github.com/users/leapcode/repos","events_url":"https://api.github.com/users/leapcode/events{/privacy}","received_events_url":"https://api.github.com/users/leapcode/received_events","type":"Organization","site_admin":false},"repo":{"id":8909650,"name":"leap_pycommon","full_name":"leapcode/leap_pycommon","owner":{"login":"leapcode","id":1700464,"avatar_url":"https://avatars.githubusercontent.com/u/1700464?v=3","gravatar_id":"","url":"https://api.github.com/users/leapcode","html_url":"https://github.com/leapcode","followers_url":"https://api.github.com/users/leapcode/followers","following_url":"https://api.github.com/users/leapcode/following{/other_user}","gists_url":"https://api.github.com/users/leapcode/gists{/gist_id}","starred_url":"https://api.github.com/users/leapcode/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/leapcode/subscriptions","organizations_url":"https://api.github.com/users/leapcode/orgs","repos_url":"https://api.github.com/users/leapcode/repos","events_url":"https://api.github.com/users/leapcode/events{/privacy}","received_events_url":"https://api.github.com/users/leapcode/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/leapcode/leap_pycommon","description":"Common utilities for leap python modules.","fork":false,"url":"https://api.github.com/repos/leapcode/leap_pycommon","forks_url":"https://api.github.com/repos/leapcode/leap_pycommon/forks","keys_url":"https://api.github.com/repos/leapcode/leap_pycommon/keys{/key_id}","collaborators_url":"https://api.github.com/repos/leapcode/leap_pycommon/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/leapcode/leap_pycommon/teams","hooks_url":"https://api.github.com/repos/leapcode/leap_pycommon/hooks","issue_events_url":"https://api.github.com/repos/leapcode/leap_pycommon/issues/events{/number}","events_url":"https://api.github.com/repos/leapcode/leap_pycommon/events","assignees_url":"https://api.github.com/repos/leapcode/leap_pycommon/assignees{/user}","branches_url":"https://api.github.com/repos/leapcode/leap_pycommon/branches{/branch}","tags_url":"https://api.github.com/repos/leapcode/leap_pycommon/tags","blobs_url":"https://api.github.com/repos/leapcode/leap_pycommon/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/leapcode/leap_pycommon/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/leapcode/leap_pycommon/git/refs{/sha}","trees_url":"https://api.github.com/repos/leapcode/leap_pycommon/git/trees{/sha}","statuses_url":"https://api.github.com/repos/leapcode/leap_pycommon/statuses/{sha}","languages_url":"https://api.github.com/repos/leapcode/leap_pycommon/languages","stargazers_url":"https://api.github.com/repos/leapcode/leap_pycommon/stargazers","contributors_url":"https://api.github.com/repos/leapcode/leap_pycommon/contributors","subscribers_url":"https://api.github.com/repos/leapcode/leap_pycommon/subscribers","subscription_url":"https://api.github.com/repos/leapcode/leap_pycommon/subscription","commits_url":"https://api.github.com/repos/leapcode/leap_pycommon/commits{/sha}","git_commits_url":"https://api.github.com/repos/leapcode/leap_pycommon/git/commits{/sha}","comments_url":"https://api.github.com/repos/leapcode/leap_pycommon/comments{/number}","issue_comment_url":"https://api.github.com/repos/leapcode/leap_pycommon/issues/comments{/number}","contents_url":"https://api.github.com/repos/leapcode/leap_pycommon/contents/{+path}","compare_url":"https://api.github.com/repos/leapcode/leap_pycommon/compare/{base}...{head}","merges_url":"https://api.github.com/repos/leapcode/leap_pycommon/merges","archive_url":"https://api.github.com/repos/leapcode/leap_pycommon/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/leapcode/leap_pycommon/downloads","issues_url":"https://api.github.com/repos/leapcode/leap_pycommon/issues{/number}","pulls_url":"https://api.github.com/repos/leapcode/leap_pycommon/pulls{/number}","milestones_url":"https://api.github.com/repos/leapcode/leap_pycommon/milestones{/number}","notifications_url":"https://api.github.com/repos/leapcode/leap_pycommon/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/leapcode/leap_pycommon/labels{/name}","releases_url":"https://api.github.com/repos/leapcode/leap_pycommon/releases{/id}","created_at":"2013-03-20T17:01:43Z","updated_at":"2015-01-26T16:41:07Z","pushed_at":"2015-01-26T20:19:55Z","git_url":"git://github.com/leapcode/leap_pycommon.git","ssh_url":"git@github.com:leapcode/leap_pycommon.git","clone_url":"https://github.com/leapcode/leap_pycommon.git","svn_url":"https://github.com/leapcode/leap_pycommon","homepage":"","size":1435,"stargazers_count":7,"watchers_count":7,"language":"Python","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":7,"mirror_url":null,"open_issues_count":2,"forks":7,"open_issues":2,"watchers":7,"default_branch":"develop"}},"_links":{"self":{"href":"https://api.github.com/repos/leapcode/leap_pycommon/pulls/95"},"html":{"href":"https://github.com/leapcode/leap_pycommon/pull/95"},"issue":{"href":"https://api.github.com/repos/leapcode/leap_pycommon/issues/95"},"comments":{"href":"https://api.github.com/repos/leapcode/leap_pycommon/issues/95/comments"},"review_comments":{"href":"https://api.github.com/repos/leapcode/leap_pycommon/pulls/95/comments"},"review_comment":{"href":"https://api.github.com/repos/leapcode/leap_pycommon/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/leapcode/leap_pycommon/pulls/95/commits"},"statuses":{"href":"https://api.github.com/repos/leapcode/leap_pycommon/statuses/801ce9890c951f6ffda011b38cec3b2f67d398b9"}}}}