Merge branch 'master' into internal

* master:
  Bump version to 0.1.1
  Add changelog
  Fix #10: Allow to manually specify ssh username and port
  Completely refactor the stream event handling
  Add missing __repr__ methods on ErrorEvent and UnhandledEvent
  Fix initialisation of error event
  Fix #11: correct handling of `identityfile` in the ssh config
  Allow example script to continue if errors are received
  Fix #9: Add a bit more detail in the documentation
  Fix #8: Support the "topic-changed" stream event
  Fix #7: Support the "reviewer-added" stream event
  Fix #6: Support the "merge-failed" stream event
  Fix #5: Move json parsing and error handling into the event factory
  Improved logging in the example script
  Fix #3: Don't treat unhandled event types as errors
  Fix #4: Keep event's raw json data in the event object
  Add __repr__ methods on event and model classes
  Remove redundant `exec_command` method
  Fix #2: Establish SSH connection in a thread-safe way
  Fix #1: Use select.select() instead of select.poll()

Change-Id: Ib91384b16acca30bfc5aadcdc1131c8bdecef583

15 files changed, 475 insertions, 105 deletions

diff --git a/HISTORY.rst b/HISTORY.rst
new file mode 100644
index 0000000..5d3e2c7
--- /dev/null
+++ b/HISTORY.rst
@@ -0,0 +1,25 @@
+.. :changelog:
+
+History
+-------
+
+0.1.1 (2013-09-13)
+++++++++++++++++++
+
+- Support for Mac OSX
+- Make the connection setup thread-safe
+- Unknown event types are no longer treated as errors
+- Clients can access event data that is not encapsulated by the event classes
+- Better handling of errors when parsing json data in the event stream
+- SSH username and port can be manually specified instead of relying on ``~/.ssh/config``
+- Support for the ``merge-failed`` event
+- Support for the ``reviewer-added`` event
+- Support for the ``topic-changed`` event
+- Add ``--verbose`` (debug logging) option in the example script
+- Add ``--ignore-stream-errors`` option in the example script
+- Improved documentation
+
+0.1.0 (2013-08-02)
+++++++++++++++++++
+
+- First released version
diff --git a/README.rst b/README.rst
index 819811f..bb0f5cd 100644
--- a/README.rst
+++ b/README.rst
@@ -1,10 +1,40 @@
 Pygerrit - Client library for interacting with Gerrit Code Review
 =================================================================
 
-Pygerrit is a Python library that can be used to process the data from events
-generated by the Gerrit Code Review system.
+Pygerrit is a Python library to interact with the
+`Gerrit Code Review`_ system over ssh.
 
-Gerrit offers a "stream-events" command that is run over ssh, and returns back
+Installation
+------------
+
+To install pygerrit, simply:
+
+.. code-block:: bash
+
+    $ pip install pygerrit
+
+
+Prerequisites
+-------------
+
+Pygerrit runs on Ubuntu 10.4 and Mac OSX 10.8.4 with Python 2.6.x and 2.7.x.
+Support for other platforms and Python versions is not guaranteed.
+
+To connect to the review server, pygerrit requires the ssh connection
+parameters (hostname, port, username) to be present in the ``.ssh/config``
+file for the current user:
+
+.. code-block:: bash
+
+    Host review
+      HostName review.example.net
+      Port 29418
+      User username
+
+Event Stream
+------------
+
+Gerrit offers a ``stream-events`` command that is run over ssh, and returns back
 a stream of events (new change uploaded, change merged, comment added, etc) as
 JSON text.
 
@@ -14,10 +44,20 @@ client to fetch them from a queue. It also allows users to easily add handling
 of custom event types, for example if they are running a customised Gerrit
 installation with non-standard events.
 
-For examples of usage, please refer to example.py.
+Refer to the `example`_ script for a brief example of how the interface
+works.
+
+
+Copyright and License
+---------------------
 
 Copyright 2011 Sony Ericsson Mobile Communications. All rights reserved.
+
 Copyright 2012 Sony Mobile Communications. All rights reserved.
 
-Licensed under The MIT License.  Please refer to the LICENSE file for full
+Licensed under The MIT License.  Please refer to the `LICENSE`_ file for full
 license details.
+
+.. _`Gerrit Code Review`: https://code.google.com/p/gerrit/
+.. _example: https://github.com/sonyxperiadev/pygerrit/blob/master/example.py
+.. _LICENSE: https://github.com/sonyxperiadev/pygerrit/blob/master/LICENSE
diff --git a/example.py b/example.py
index 8fa99a7..904f690 100755
--- a/example.py
+++ b/example.py
@@ -33,7 +33,7 @@ import time
 
 from pygerrit.client import GerritClient
 from pygerrit.error import GerritError
-from pygerrit.stream import GerritStreamErrorEvent
+from pygerrit.events import ErrorEvent
 
 
 def _main():
@@ -42,6 +42,11 @@ def _main():
     parser.add_option('-g', '--gerrit-hostname', dest='hostname',
                       default='review',
                       help='gerrit server hostname (default: %default)')
+    parser.add_option('-p', '--port', dest='port',
+                      type='int', default=29418,
+                      help='port number (default: %default)')
+    parser.add_option('-u', '--username', dest='username',
+                      help='username')
     parser.add_option('-b', '--blocking', dest='blocking',
                       action='store_true',
                       help='block on event get (default: False)')
@@ -49,15 +54,25 @@ def _main():
                       default=None, type='int',
                       help='timeout (seconds) for blocking event get '
                            '(default: None)')
+    parser.add_option('-v', '--verbose', dest='verbose',
+                      action='store_true',
+                      help='enable verbose (debug) logging')
+    parser.add_option('-i', '--ignore-stream-errors', dest='ignore',
+                      action='store_true',
+                      help='do not exit when an error event is received')
 
     (options, _args) = parser.parse_args()
     if options.timeout and not options.blocking:
         parser.error('Can only use --timeout with --blocking')
 
-    logging.basicConfig(format='%(message)s', level=logging.INFO)
+    level = logging.DEBUG if options.verbose else logging.INFO
+    logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s',
+                        level=level)
 
     try:
-        gerrit = GerritClient(host=options.hostname)
+        gerrit = GerritClient(host=options.hostname,
+                              username=options.username,
+                              port=options.port)
         logging.info("Connected to Gerrit version [%s]",
                      gerrit.gerrit_version())
         gerrit.start_event_stream()
@@ -71,8 +86,8 @@ def _main():
             event = gerrit.get_event(block=options.blocking,
                                      timeout=options.timeout)
             if event:
-                logging.info("Event: %s", str(event))
-                if isinstance(event, GerritStreamErrorEvent):
+                logging.info("Event: %s", event)
+                if isinstance(event, ErrorEvent) and not options.ignore:
                     logging.error(event.error)
                     errors.set()
                     break
@@ -83,6 +98,7 @@ def _main():
     except KeyboardInterrupt:
         logging.info("Terminated by user")
     finally:
+        logging.debug("Stopping event stream...")
         gerrit.stop_event_stream()
 
     if errors.isSet():
diff --git a/pygerrit/__init__.py b/pygerrit/__init__.py
index 98d0f92..76c8321 100644
--- a/pygerrit/__init__.py
+++ b/pygerrit/__init__.py
@@ -22,7 +22,7 @@
 
 """ Module to interface with Gerrit. """
 
-__numversion__ = (0, 1, 0)
+__numversion__ = (0, 1, 1)
 __version__ = '.'.join([str(num) for num in __numversion__])
 
 
diff --git a/pygerrit/client.py b/pygerrit/client.py
index b6502e1..456ce2e 100644
--- a/pygerrit/client.py
+++ b/pygerrit/client.py
@@ -37,11 +37,11 @@ class GerritClient(object):
 
     """ Gerrit client interface. """
 
-    def __init__(self, host):
+    def __init__(self, host, username=None, port=None):
         self._factory = GerritEventFactory()
         self._events = Queue()
         self._stream = None
-        self._ssh_client = GerritSSHClient(host)
+        self._ssh_client = GerritSSHClient(host, username=username, port=port)
 
     def gerrit_version(self):
         """ Return the version of Gerrit that is connected to. """
@@ -112,15 +112,15 @@ class GerritClient(object):
         except Empty:
             return None
 
-    def put_event(self, json_data):
-        """ Create event from `json_data` and add it to the queue.
+    def put_event(self, data):
+        """ Create event from `data` and add it to the queue.
 
         Raise GerritError if the queue is full, or the factory could not
         create the event.
 
         """
         try:
-            event = self._factory.create(json_data)
+            event = self._factory.create(data)
             self._events.put(event)
         except Full:
             raise GerritError("Unable to add event: queue is full")
diff --git a/pygerrit/events.py b/pygerrit/events.py
index 8a5c21f..ebbb327 100644
--- a/pygerrit/events.py
+++ b/pygerrit/events.py
@@ -23,6 +23,9 @@
 
 """ Gerrit event classes. """
 
+import json
+import logging
+
 from .error import GerritError
 from .models import Account, Approval, Change, Patchset, RefUpdate
 
@@ -53,19 +56,27 @@ class GerritEventFactory(object):
         return decorate
 
     @classmethod
-    def create(cls, json_data):
+    def create(cls, data):
         """ Create a new event instance.
 
-        Return an instance of the `GerritEvent` subclass from `json_data`
-        Raise GerritError if `json_data` does not contain a `type` key, or
-        no corresponding event is registered.
+        Return an instance of the `GerritEvent` subclass after converting
+        `data` to json.
+
+        Raise GerritError if json parsed from `data` does not contain a `type`
+        key.
 
         """
+        try:
+            json_data = json.loads(data)
+        except ValueError as err:
+            logging.debug("Failed to load json data: %s: [%s]", str(err), data)
+            json_data = json.loads(ErrorEvent.error_json(err))
+
         if not "type" in json_data:
             raise GerritError("`type` not in json_data")
         name = json_data["type"]
         if not name in cls._events:
-            raise GerritError("Unknown event: %s" % name)
+            name = 'unhandled-event'
         event = cls._events[name]
         module_name = event[0]
         class_name = event[1]
@@ -78,11 +89,39 @@ class GerritEvent(object):
 
     """ Gerrit event base class. """
 
-    def __init__(self):
-        pass
+    def __init__(self, json_data):
+        self.json = json_data
+
+
+@GerritEventFactory.register("unhandled-event")
+class UnhandledEvent(GerritEvent):
+
+    """ Unknown event type received in json data from Gerrit's event stream. """
+
+    def __init__(self, json_data):
+        super(UnhandledEvent, self).__init__(json_data)
+
+    def __repr__(self):
+        return u"<UnhandledEvent>"
+
+
+@GerritEventFactory.register("error-event")
+class ErrorEvent(GerritEvent):
+
+    """ Error occurred when processing json data from Gerrit's event stream. """
 
-    def __str__(self):
-        return u"%s" % self.name  # pylint: disable=no-member
+    def __init__(self, json_data):
+        super(ErrorEvent, self).__init__(json_data)
+        self.error = json_data["error"]
+
+    @classmethod
+    def error_json(cls, error):
+        """ Return a json string for the `error`. """
+        return '{"type":"error-event",' \
+               '"error":"%s"}' % str(error)
+
+    def __repr__(self):
+        return u"<ErrorEvent: %s>" % self.error
 
 
 @GerritEventFactory.register("patchset-created")
@@ -91,7 +130,7 @@ class PatchsetCreatedEvent(GerritEvent):
     """ Gerrit "patchset-created" event. """
 
     def __init__(self, json_data):
-        super(PatchsetCreatedEvent, self).__init__()
+        super(PatchsetCreatedEvent, self).__init__(json_data)
         try:
             self.change = Change(json_data["change"])
             self.patchset = Patchset(json_data["patchSet"])
@@ -99,6 +138,11 @@ class PatchsetCreatedEvent(GerritEvent):
         except KeyError as e:
             raise GerritError("PatchsetCreatedEvent: %s" % e)
 
+    def __repr__(self):
+        return u"<PatchsetCreatedEvent>: %s %s %s" % (self.change,
+                                                      self.patchset,
+                                                      self.uploader)
+
 
 @GerritEventFactory.register("draft-published")
 class DraftPublishedEvent(GerritEvent):
@@ -106,7 +150,7 @@ class DraftPublishedEvent(GerritEvent):
     """ Gerrit "draft-published" event. """
 
     def __init__(self, json_data):
-        super(DraftPublishedEvent, self).__init__()
+        super(DraftPublishedEvent, self).__init__(json_data)
         try:
             self.change = Change(json_data["change"])
             self.patchset = Patchset(json_data["patchSet"])
@@ -114,6 +158,11 @@ class DraftPublishedEvent(GerritEvent):
         except KeyError as e:
             raise GerritError("DraftPublishedEvent: %s" % e)
 
+    def __repr__(self):
+        return u"<DraftPublishedEvent>: %s %s %s" % (self.change,
+                                                     self.patchset,
+                                                     self.uploader)
+
 
 @GerritEventFactory.register("comment-added")
 class CommentAddedEvent(GerritEvent):
@@ -121,7 +170,7 @@ class CommentAddedEvent(GerritEvent):
     """ Gerrit "comment-added" event. """
 
     def __init__(self, json_data):
-        super(CommentAddedEvent, self).__init__()
+        super(CommentAddedEvent, self).__init__(json_data)
         try:
             self.change = Change(json_data["change"])
             self.patchset = Patchset(json_data["patchSet"])
@@ -134,6 +183,11 @@ class CommentAddedEvent(GerritEvent):
         except (KeyError, ValueError) as e:
             raise GerritError("CommentAddedEvent: %s" % e)
 
+    def __repr__(self):
+        return u"<CommentAddedEvent>: %s %s %s" % (self.change,
+                                                   self.patchset,
+                                                   self.author)
+
 
 @GerritEventFactory.register("change-merged")
 class ChangeMergedEvent(GerritEvent):
@@ -141,7 +195,7 @@ class ChangeMergedEvent(GerritEvent):
     """ Gerrit "change-merged" event. """
 
     def __init__(self, json_data):
-        super(ChangeMergedEvent, self).__init__()
+        super(ChangeMergedEvent, self).__init__(json_data)
         try:
             self.change = Change(json_data["change"])
             self.patchset = Patchset(json_data["patchSet"])
@@ -149,6 +203,32 @@ class ChangeMergedEvent(GerritEvent):
         except KeyError as e:
             raise GerritError("ChangeMergedEvent: %s" % e)
 
+    def __repr__(self):
+        return u"<ChangeMergedEvent>: %s %s %s" % (self.change,
+                                                   self.patchset,
+                                                   self.submitter)
+
+
+@GerritEventFactory.register("merge-failed")
+class MergeFailedEvent(GerritEvent):
+
+    """ Gerrit "merge-failed" event. """
+
+    def __init__(self, json_data):
+        super(MergeFailedEvent, self).__init__(json_data)
+        try:
+            self.change = Change(json_data["change"])
+            self.patchset = Patchset(json_data["patchSet"])
+            self.submitter = Account(json_data["submitter"])
+            self.reason = json_data["reason"]
+        except KeyError as e:
+            raise GerritError("MergeFailedEvent: %s" % e)
+
+    def __repr__(self):
+        return u"<MergeFailedEvent>: %s %s %s" % (self.change,
+                                                  self.patchset,
+                                                  self.submitter)
+
 
 @GerritEventFactory.register("change-abandoned")
 class ChangeAbandonedEvent(GerritEvent):
@@ -156,7 +236,7 @@ class ChangeAbandonedEvent(GerritEvent):
     """ Gerrit "change-abandoned" event. """
 
     def __init__(self, json_data):
-        super(ChangeAbandonedEvent, self).__init__()
+        super(ChangeAbandonedEvent, self).__init__(json_data)
         try:
             self.change = Change(json_data["change"])
             self.patchset = Patchset.from_json(json_data)
@@ -165,6 +245,11 @@ class ChangeAbandonedEvent(GerritEvent):
         except KeyError as e:
             raise GerritError("ChangeAbandonedEvent: %s" % e)
 
+    def __repr__(self):
+        return u"<ChangeAbandonedEvent>: %s %s %s" % (self.change,
+                                                      self.patchset,
+                                                      self.abandoner)
+
 
 @GerritEventFactory.register("change-restored")
 class ChangeRestoredEvent(GerritEvent):
@@ -172,7 +257,7 @@ class ChangeRestoredEvent(GerritEvent):
     """ Gerrit "change-restored" event. """
 
     def __init__(self, json_data):
-        super(ChangeRestoredEvent, self).__init__()
+        super(ChangeRestoredEvent, self).__init__(json_data)
         try:
             self.change = Change(json_data["change"])
             self.patchset = Patchset.from_json(json_data)
@@ -181,6 +266,11 @@ class ChangeRestoredEvent(GerritEvent):
         except KeyError as e:
             raise GerritError("ChangeRestoredEvent: %s" % e)
 
+    def __repr__(self):
+        return u"<ChangeRestoredEvent>: %s %s %s" % (self.change,
+                                                     self.patchset,
+                                                     self.restorer)
+
 
 @GerritEventFactory.register("ref-updated")
 class RefUpdatedEvent(GerritEvent):
@@ -188,9 +278,55 @@ class RefUpdatedEvent(GerritEvent):
     """ Gerrit "ref-updated" event. """
 
     def __init__(self, json_data):
-        super(RefUpdatedEvent, self).__init__()
+        super(RefUpdatedEvent, self).__init__(json_data)
         try:
             self.ref_update = RefUpdate(json_data["refUpdate"])
             self.submitter = Account.from_json(json_data, "submitter")
         except KeyError as e:
             raise GerritError("RefUpdatedEvent: %s" % e)
+
+    def __repr__(self):
+        return u"<RefUpdatedEvent>: %s %s" % (self.ref_update, self.submitter)
+
+
+@GerritEventFactory.register("reviewer-added")
+class ReviewerAddedEvent(GerritEvent):
+
+    """ Gerrit "reviewer-added" event. """
+
+    def __init__(self, json_data):
+        super(ReviewerAddedEvent, self).__init__(json_data)
+        try:
+            self.change = Change(json_data["change"])
+            self.patchset = Patchset.from_json(json_data)
+            self.reviewer = Account(json_data["reviewer"])
+        except KeyError as e:
+            raise GerritError("ReviewerAddedEvent: %s" % e)
+
+    def __repr__(self):
+        return u"<ReviewerAddedEvent>: %s %s %s" % (self.change,
+                                                    self.patchset,
+                                                    self.reviewer)
+
+
+@GerritEventFactory.register("topic-changed")
+class TopicChangedEvent(GerritEvent):
+
+    """ Gerrit "topic-changed" event. """
+
+    def __init__(self, json_data):
+        super(TopicChangedEvent, self).__init__(json_data)
+        try:
+            self.change = Change(json_data["change"])
+            self.changer = Account(json_data["changer"])
+            if "oldTopic" in json_data:
+                self.oldtopic = json_data["oldTopic"]
+            else:
+                self.oldtopic = ""
+        except KeyError as e:
+            raise GerritError("TopicChangedEvent: %s" % e)
+
+    def __repr__(self):
+        return u"<TopicChangedEvent>: %s %s [%s]" % (self.change,
+                                                     self.changer,
+                                                     self.oldtopic)
diff --git a/pygerrit/models.py b/pygerrit/models.py
index 2e79949..ac9575a 100644
--- a/pygerrit/models.py
+++ b/pygerrit/models.py
@@ -34,6 +34,10 @@ class Account(object):
         self.name = from_json(json_data, "name")
         self.email = from_json(json_data, "email")
 
+    def __repr__(self):
+        return u"<Account %s%s>" % (self.name,
+                                    " (%s)" % self.email if self.email else "")
+
     @staticmethod
     def from_json(json_data, key):
         """ Create an Account instance.
@@ -61,6 +65,9 @@ class Change(object):
         self.url = from_json(json_data, "url")
         self.owner = Account.from_json(json_data, "owner")
 
+    def __repr__(self):
+        return u"<Change %s, %s, %s>" % (self.number, self.project, self.branch)
+
 
 class Patchset(object):
 
@@ -72,6 +79,9 @@ class Patchset(object):
         self.ref = from_json(json_data, "ref")
         self.uploader = Account.from_json(json_data, "uploader")
 
+    def __repr__(self):
+        return u"<Patchset %s, %s>" % (self.number, self.revision)
+
     @staticmethod
     def from_json(json_data):
         r""" Create a Patchset instance.
@@ -94,6 +104,9 @@ class Approval(object):
         self.value = from_json(json_data, "value")
         self.description = from_json(json_data, "description")
 
+    def __repr__(self):
+        return u"<Approval %s %s>" % (self.description, self.value)
+
 
 class RefUpdate(object):
 
@@ -104,3 +117,7 @@ class RefUpdate(object):
         self.newrev = from_json(json_data, "newRev")
         self.refname = from_json(json_data, "refName")
         self.project = from_json(json_data, "project")
+
+    def __repr__(self):
+        return "<RefUpdate %s %s %s %s>" % \
+            (self.project, self.refname, self.oldrev, self.newrev)
diff --git a/pygerrit/ssh.py b/pygerrit/ssh.py
index 3e2add1..e779715 100644
--- a/pygerrit/ssh.py
+++ b/pygerrit/ssh.py
@@ -25,6 +25,7 @@
 from os.path import abspath, expanduser, isfile
 import re
 import socket
+from threading import Event, Lock
 
 from .error import GerritError
 
@@ -64,18 +65,19 @@ class GerritSSHClient(SSHClient):
 
     """ Gerrit SSH Client, wrapping the paramiko SSH Client. """
 
-    def __init__(self, hostname):
+    def __init__(self, hostname, username=None, port=None):
         """ Initialise and connect to SSH. """
         super(GerritSSHClient, self).__init__()
         self.remote_version = None
         self.hostname = hostname
-        self.connected = False
-
-    def _connect(self):
-        """ Connect to the remote if not already connected. """
-        if self.connected:
-            return
-        self.load_system_host_keys()
+        self.username = username
+        self.key_filename = None
+        self.port = port
+        self.connected = Event()
+        self.lock = Lock()
+
+    def _configure(self):
+        """ Configure the ssh parameters from the config file. """
         configfile = expanduser("~/.ssh/config")
         if not isfile(configfile):
             raise GerritError("ssh config file '%s' does not exist" %
@@ -88,22 +90,29 @@ class GerritSSHClient(SSHClient):
             raise GerritError("No ssh config for host %s" % self.hostname)
         if not 'hostname' in data or not 'port' in data or not 'user' in data:
             raise GerritError("Missing configuration data in %s" % configfile)
-        key_filename = None
+        self.hostname = data['hostname']
+        self.username = data['user']
         if 'identityfile' in data:
-            key_filename = abspath(expanduser(data['identityfile']))
+            key_filename = abspath(expanduser(data['identityfile'][0]))
             if not isfile(key_filename):
                 raise GerritError("Identity file '%s' does not exist" %
                                   key_filename)
+            self.key_filename = key_filename
         try:
-            port = int(data['port'])
+            self.port = int(data['port'])
         except ValueError:
             raise GerritError("Invalid port: %s" % data['port'])
+
+    def _do_connect(self):
+        """ Connect to the remote. """
+        self.load_system_host_keys()
+        if self.username is None or self.port is None:
+            self._configure()
         try:
-            self.connect(hostname=data['hostname'],
-                         port=port,
-                         username=data['user'],
-                         key_filename=key_filename)
-            self.connected = True
+            self.connect(hostname=self.hostname,
+                         port=self.port,
+                         username=self.username,
+                         key_filename=self.key_filename)
         except socket.error as e:
             raise GerritError("Failed to connect to server: %s" % e)
 
@@ -114,17 +123,20 @@ class GerritSSHClient(SSHClient):
         except AttributeError:
             self.remote_version = None
 
-    def exec_command(self, command, bufsize=1, timeout=None, get_pty=False):
-        """ Execute the command.
-
-        Make sure we're connected and then execute the command.
-
-        Return a tuple of stdin, stdout, stderr.
-
-        """
-        self._connect()
-        return super(GerritSSHClient, self).\
-            exec_command(command, bufsize, timeout, get_pty)
+    def _connect(self):
+        """ Connect to the remote if not already connected. """
+        if not self.connected.is_set():
+            try:
+                self.lock.acquire()
+                # Another thread may have connected while we were
+                # waiting to acquire the lock
+                if not self.connected.is_set():
+                    self._do_connect()
+                    self.connected.set()
+            except GerritError:
+                raise
+            finally:
+                self.lock.release()
 
     def get_remote_version(self):
         """ Return the version of the remote Gerrit server. """
@@ -138,17 +150,24 @@ class GerritSSHClient(SSHClient):
     def run_gerrit_command(self, command):
         """ Run the given command.
 
-        Run `command` and return a `GerritSSHCommandResult`.
+        Make sure we're connected to the remote server, and run `command`.
+
+        Return the results as a `GerritSSHCommandResult`.
 
-        Raise `ValueError` if `command` is not a string.
+        Raise `ValueError` if `command` is not a string, or `GerritError` if
+        command execution fails.
 
         """
         if not isinstance(command, basestring):
             raise ValueError("command must be a string")
         gerrit_command = "gerrit " + command
 
+        self._connect()
         try:
-            stdin, stdout, stderr = self.exec_command(gerrit_command)
+            stdin, stdout, stderr = self.exec_command(gerrit_command,
+                                                      bufsize=1,
+                                                      timeout=None,
+                                                      get_pty=False)
         except SSHException as err:
             raise GerritError("Command execution error: %s" % err)
         return GerritSSHCommandResult(command, stdin, stdout, stderr)
diff --git a/pygerrit/stream.py b/pygerrit/stream.py
index d7633e8..37b09f1 100644
--- a/pygerrit/stream.py
+++ b/pygerrit/stream.py
@@ -26,23 +26,9 @@ Class to listen to the Gerrit event stream and dispatch events.
 
 """
 
-import json
-import logging
-from select import poll, POLLIN
 from threading import Thread, Event
 
-from .error import GerritError
-from .events import GerritEvent, GerritEventFactory
-
-
-@GerritEventFactory.register("gerrit-stream-error")
-class GerritStreamErrorEvent(GerritEvent):
-
-    """ Represents an error when handling the gerrit event stream. """
-
-    def __init__(self, json_data):
-        super(GerritStreamErrorEvent, self).__init__()
-        self.error = json_data["error"]
+from .events import ErrorEvent
 
 
 class GerritStream(Thread):
@@ -62,29 +48,22 @@ class GerritStream(Thread):
 
     def _error_event(self, error):
         """ Dispatch `error` to the Gerrit client. """
-        json_data = json.loads('{"type":"gerrit-stream-error",'
-                               '"error":"%s"}' % str(error))
-        self._gerrit.put_event(json_data)
+        self._gerrit.put_event(ErrorEvent.error_json(error))
 
     def run(self):
         """ Listen to the stream and send events to the client. """
-        try:
-            result = self._ssh_client.run_gerrit_command("stream-events")
-        except GerritError as e:
-            self._error_event(e)
-        else:
-            poller = poll()
-            stdout = result.stdout
-            poller.register(stdout.channel)
-            while not self._stop.is_set():
-                data = poller.poll()
-                for (handle, event) in data:
-                    if handle == stdout.channel.fileno() and event == POLLIN:
-                        try:
-                            line = stdout.readline()
-                            json_data = json.loads(line)
-                            self._gerrit.put_event(json_data)
-                        except (ValueError, IOError) as err:
-                            self._error_event(err)
-                        except GerritError as err:
-                            logging.error("Failed to put event: %s", err)
+        channel = self._ssh_client.get_transport().open_session()
+        channel.exec_command("gerrit stream-events")
+        stdout = channel.makefile()
+        stderr = channel.makefile_stderr()
+        while not self._stop.is_set():
+            if channel.exit_status_ready():
+                if channel.recv_stderr_ready():
+                    error = stderr.readline().strip()
+                else:
+                    error = "Remote server connection closed"
+                self._error_event(error)
+                self._stop.set()
+            elif channel.recv_ready():
+                data = stdout.readline()
+                self._gerrit.put_event(data)
diff --git a/testdata/invalid-json.txt b/testdata/invalid-json.txt
new file mode 100644
index 0000000..f7a60bb
--- /dev/null
+++ b/testdata/invalid-json.txt
@@ -0,0 +1,4 @@
+)]}'
+{"type":"user-defined-event",
+ "title":"Event title",
+ "description":"Event description"}
diff --git a/testdata/merge-failed-event.txt b/testdata/merge-failed-event.txt
new file mode 100644
index 0000000..2d29d29
--- /dev/null
+++ b/testdata/merge-failed-event.txt
@@ -0,0 +1,19 @@
+{"type":"merge-failed",
+ "change":{"project":"project-name",
+           "branch":"branch-name",
+           "topic":"topic-name",
+           "id":"Ideadbeefdeadbeefdeadbeefdeadbeefdeadbeef",
+           "number":"123456",
+           "subject":"Commit message subject",
+           "owner":{"name":"Owner Name",
+                    "email":"owner@example.com"},
+           "url":"http://review.example.com/123456"},
+ "patchSet":{"number":"4",
+             "revision":"deadbeefdeadbeefdeadbeefdeadbeefdeadbeef",
+             "ref":"refs/changes/56/123456/4",
+             "uploader":{"name":"Uploader Name",
+                         "email":"uploader@example.com"},
+             "createdOn":1341370514},
+ "submitter":{"name":"Submitter Name",
+              "email":"submitter@example.com"},
+ "reason":"Merge failed reason"}
diff --git a/testdata/reviewer-added-event.txt b/testdata/reviewer-added-event.txt
new file mode 100644
index 0000000..b460afc
--- /dev/null
+++ b/testdata/reviewer-added-event.txt
@@ -0,0 +1,18 @@
+{"type":"reviewer-added",
+ "change":{"project":"project-name",
+           "branch":"branch-name",
+           "topic":"topic-name",
+           "id":"Ideadbeefdeadbeefdeadbeefdeadbeefdeadbeef",
+           "number":"123456",
+           "subject":"Commit message subject",
+           "owner":{"name":"Owner Name",
+                    "email":"owner@example.com"},
+           "url":"http://review.example.com/123456"},
+ "patchSet":{"number":"4",
+             "revision":"deadbeefdeadbeefdeadbeefdeadbeefdeadbeef",
+             "ref":"refs/changes/56/123456/4",
+             "uploader":{"name":"Uploader Name",
+                         "email":"uploader@example.com"},
+             "createdOn":1341370514},
+ "reviewer":{"name":"Reviewer Name",
+             "email":"reviewer@example.com"}}
diff --git a/testdata/topic-changed-event.txt b/testdata/topic-changed-event.txt
new file mode 100644
index 0000000..1847440
--- /dev/null
+++ b/testdata/topic-changed-event.txt
@@ -0,0 +1,13 @@
+{"type":"topic-changed",
+ "change":{"project":"project-name",
+           "branch":"branch-name",
+           "topic":"topic-name",
+           "id":"Ideadbeefdeadbeefdeadbeefdeadbeefdeadbeef",
+           "number":"123456",
+           "subject":"Commit message subject",
+           "owner":{"name":"Owner Name",
+                    "email":"owner@example.com"},
+           "url":"http://review.example.com/123456"},
+ "changer":{"name":"Changer Name",
+            "email":"changer@example.com"},
+ "oldTopic":"old-topic"}
diff --git a/testdata/unhandled-event.txt b/testdata/unhandled-event.txt
new file mode 100644
index 0000000..6824cc8
--- /dev/null
+++ b/testdata/unhandled-event.txt
@@ -0,0 +1,3 @@
+{"type":"this-event-is-not-handled",
+ "title":"Unhandled event title",
+ "description":"Unhandled event description"}
diff --git a/unittests.py b/unittests.py
index 4b7c2b8..fac78e1 100755
--- a/unittests.py
+++ b/unittests.py
@@ -32,7 +32,8 @@ import unittest
 from pygerrit.events import PatchsetCreatedEvent, \
     RefUpdatedEvent, ChangeMergedEvent, CommentAddedEvent, \
     ChangeAbandonedEvent, ChangeRestoredEvent, \
-    DraftPublishedEvent, GerritEventFactory, GerritEvent
+    DraftPublishedEvent, GerritEventFactory, GerritEvent, UnhandledEvent, \
+    ErrorEvent, MergeFailedEvent, ReviewerAddedEvent, TopicChangedEvent
 from pygerrit.client import GerritClient
 from setup import REQUIRES as setup_requires
 
@@ -43,7 +44,7 @@ class UserDefinedEvent(GerritEvent):
     """ Dummy event class to test event registration. """
 
     def __init__(self, json_data):
-        super(UserDefinedEvent, self).__init__()
+        super(UserDefinedEvent, self).__init__(json_data)
         self.title = json_data['title']
         self.description = json_data['description']
 
@@ -55,9 +56,10 @@ def _create_event(name, gerrit):
     data, then add as an event in the `gerrit` client.
 
     """
-    data = open(os.path.join("testdata", name + ".txt"))
-    json_data = json.loads(data.read().replace("\n", ""))
-    gerrit.put_event(json_data)
+    testfile = open(os.path.join("testdata", name + ".txt"))
+    data = testfile.read().replace("\n", "")
+    gerrit.put_event(data)
+    return data
 
 
 class TestConsistentDependencies(unittest.TestCase):
@@ -171,6 +173,31 @@ class TestGerritEvents(unittest.TestCase):
         self.assertEquals(event.submitter.name, "Submitter Name")
         self.assertEquals(event.submitter.email, "submitter@example.com")
 
+    def test_merge_failed(self):
+        _create_event("merge-failed-event", self.gerrit)
+        event = self.gerrit.get_event(False)
+        self.assertTrue(isinstance(event, MergeFailedEvent))
+        self.assertEquals(event.name, "merge-failed")
+        self.assertEquals(event.change.project, "project-name")
+        self.assertEquals(event.change.branch, "branch-name")
+        self.assertEquals(event.change.topic, "topic-name")
+        self.assertEquals(event.change.change_id,
+                          "Ideadbeefdeadbeefdeadbeefdeadbeefdeadbeef")
+        self.assertEquals(event.change.number, "123456")
+        self.assertEquals(event.change.subject, "Commit message subject")
+        self.assertEquals(event.change.url, "http://review.example.com/123456")
+        self.assertEquals(event.change.owner.name, "Owner Name")
+        self.assertEquals(event.change.owner.email, "owner@example.com")
+        self.assertEquals(event.patchset.number, "4")
+        self.assertEquals(event.patchset.revision,
+                          "deadbeefdeadbeefdeadbeefdeadbeefdeadbeef")
+        self.assertEquals(event.patchset.ref, "refs/changes/56/123456/4")
+        self.assertEquals(event.patchset.uploader.name, "Uploader Name")
+        self.assertEquals(event.patchset.uploader.email, "uploader@example.com")
+        self.assertEquals(event.submitter.name, "Submitter Name")
+        self.assertEquals(event.submitter.email, "submitter@example.com")
+        self.assertEquals(event.reason, "Merge failed reason")
+
     def test_comment_added(self):
         _create_event("comment-added-event", self.gerrit)
         event = self.gerrit.get_event(False)
@@ -202,6 +229,30 @@ class TestGerritEvents(unittest.TestCase):
         self.assertEquals(event.author.name, "Author Name")
         self.assertEquals(event.author.email, "author@example.com")
 
+    def test_reviewer_added(self):
+        _create_event("reviewer-added-event", self.gerrit)
+        event = self.gerrit.get_event(False)
+        self.assertTrue(isinstance(event, ReviewerAddedEvent))
+        self.assertEquals(event.name, "reviewer-added")
+        self.assertEquals(event.change.project, "project-name")
+        self.assertEquals(event.change.branch, "branch-name")
+        self.assertEquals(event.change.topic, "topic-name")
+        self.assertEquals(event.change.change_id,
+                          "Ideadbeefdeadbeefdeadbeefdeadbeefdeadbeef")
+        self.assertEquals(event.change.number, "123456")
+        self.assertEquals(event.change.subject, "Commit message subject")
+        self.assertEquals(event.change.url, "http://review.example.com/123456")
+        self.assertEquals(event.change.owner.name, "Owner Name")
+        self.assertEquals(event.change.owner.email, "owner@example.com")
+        self.assertEquals(event.patchset.number, "4")
+        self.assertEquals(event.patchset.revision,
+                          "deadbeefdeadbeefdeadbeefdeadbeefdeadbeef")
+        self.assertEquals(event.patchset.ref, "refs/changes/56/123456/4")
+        self.assertEquals(event.patchset.uploader.name, "Uploader Name")
+        self.assertEquals(event.patchset.uploader.email, "uploader@example.com")
+        self.assertEquals(event.reviewer.name, "Reviewer Name")
+        self.assertEquals(event.reviewer.email, "reviewer@example.com")
+
     def test_change_abandoned(self):
         _create_event("change-abandoned-event", self.gerrit)
         event = self.gerrit.get_event(False)
@@ -240,6 +291,25 @@ class TestGerritEvents(unittest.TestCase):
         self.assertEquals(event.restorer.email, "restorer@example.com")
         self.assertEquals(event.reason, "Restore reason")
 
+    def test_topic_changed(self):
+        _create_event("topic-changed-event", self.gerrit)
+        event = self.gerrit.get_event(False)
+        self.assertTrue(isinstance(event, TopicChangedEvent))
+        self.assertEquals(event.name, "topic-changed")
+        self.assertEquals(event.change.project, "project-name")
+        self.assertEquals(event.change.branch, "branch-name")
+        self.assertEquals(event.change.topic, "topic-name")
+        self.assertEquals(event.change.change_id,
+                          "Ideadbeefdeadbeefdeadbeefdeadbeefdeadbeef")
+        self.assertEquals(event.change.number, "123456")
+        self.assertEquals(event.change.subject, "Commit message subject")
+        self.assertEquals(event.change.url, "http://review.example.com/123456")
+        self.assertEquals(event.change.owner.name, "Owner Name")
+        self.assertEquals(event.change.owner.email, "owner@example.com")
+        self.assertEquals(event.changer.name, "Changer Name")
+        self.assertEquals(event.changer.email, "changer@example.com")
+        self.assertEquals(event.oldtopic, "old-topic")
+
     def test_user_defined_event(self):
         _create_event("user-defined-event", self.gerrit)
         event = self.gerrit.get_event(False)
@@ -247,6 +317,17 @@ class TestGerritEvents(unittest.TestCase):
         self.assertEquals(event.title, "Event title")
         self.assertEquals(event.description, "Event description")
 
+    def test_unhandled_event(self):
+        data = _create_event("unhandled-event", self.gerrit)
+        event = self.gerrit.get_event(False)
+        self.assertTrue(isinstance(event, UnhandledEvent))
+        self.assertEquals(event.json, json.loads(data))
+
+    def test_invalid_json(self):
+        _create_event("invalid-json", self.gerrit)
+        event = self.gerrit.get_event(False)
+        self.assertTrue(isinstance(event, ErrorEvent))
+
     def test_add_duplicate_event(self):
         try:
             @GerritEventFactory.register("user-defined-event")