6 files changed, 264 insertions, 101 deletions

diff --git a/README.rst b/README.rst
index d13f90a..66696b8 100644
--- a/README.rst
+++ b/README.rst
@@ -35,7 +35,7 @@ Mastodon.py
    mastodon.toot('Tooting from python using #mastodonpy !')
 
 Python wrapper for the Mastodon ( https://github.com/tootsuite/mastodon/ ) API. 
-Feature complete for public API as of version v2.0 and easy to get started with.
+Feature complete for public API as of Mastodon version 2.0.0 and easy to get started with.
 
 You can install Mastodon.py via pypi:
 
diff --git a/docs/index.rst b/docs/index.rst
index 14ec49f..fa021d2 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -37,7 +37,7 @@ Mastodon.py
    )
    mastodon.toot('Tooting from python using #mastodonpy !')
 
-`Mastodon`_ is an ostatus based twitter-like federated social 
+`Mastodon`_ is an ActivityPub and OStatus based twitter-like federated social 
 network node. It has an API that allows you to interact with its 
 every aspect. This is a simple python wrapper for that api, provided
 as a single python module. By default, it talks to the 
@@ -112,23 +112,23 @@ Error handling
 When Mastodon.py encounters an error, it will raise an exception, generally with
 some text included to tell you what went wrong. 
 
-The base class that all mastodon exceptions inherit from is the MastodonError
-class. If you are only interested in the fact an error was raised somewhere in
+The base class that all mastodon exceptions inherit from is `MastodonError`. 
+If you are only interested in the fact an error was raised somewhere in
 Mastodon.py, and not the details, this is the exception you can catch.
 
-MastodonIllegalArgumentError is generally a programming problem - you asked the 
+`MastodonIllegalArgumentError` is generally a programming problem - you asked the 
 API to do something obviously invalid (i.e. specify a privacy scope that does
 not exist).
 
-MastodonFileNotFoundError and MastodonNetworkError are IO errors - could be you
+`MastodonFileNotFoundError` and `MastodonNetworkError` are IO errors - could be you
 specified a wrong URL, could be the internet is down or your hard drive is
 dying. They inherit from MastodonIOError, for easy catching.
 
-MastodonAPIError is an error returned from the Mastodon instance - the server
+`MastodonAPIError` is an error returned from the Mastodon instance - the server
 has decided it can't fullfill your request (i.e. you requested info on a user that
 does not exist).
 
-MastodonRatelimitError is raised when you hit an API rate limit. You should try 
+`MastodonRatelimitError` is raised when you hit an API rate limit. You should try 
 again after a while (see the rate limiting section above).
 
 Return values
@@ -172,10 +172,10 @@ Toot dicts
         'uri': # Descriptor for the toot
             # EG 'tag:mastodon.social,2016-11-25:objectId=<id>:objectType=Status'
         'url': # URL of the toot
-        'account': # Account dict for the account which posted the status
+        'account': # User dict for the account which posted the status
         'in_reply_to_id': # Numerical id of the toot this toot is in response to
         'in_reply_to_account_id': # Numerical id of the account this toot is in response to
-        'reblog': # Denotes whether the toot is a reblog
+        'reblog': # Denotes whether the toot is a reblog. If so, set to the original toot dict.
         'content': # Content of the toot, as HTML: '<p>Hello from Python</p>'
         'created_at': # Creation time
         'reblogs_count': # Number of reblogs
@@ -531,10 +531,35 @@ Streaming
 ---------
 These functions allow access to the streaming API.
 
-.. automethod:: Mastodon.user_stream
-.. automethod:: Mastodon.public_stream
-.. automethod:: Mastodon.hashtag_stream
+If async is False, these  methods block forever (or until an
+exception is raised).
 
+If async is True, the listener will listen on another thread and these methods
+will return a handle corresponding to the open connection. The
+connection may be closed at any time by calling its close() method.
+
+The streaming functions take instances of `StreamListener` as a parameter.
+A `CallbackStreamListener` class that allows you to specify function callbacks 
+directly is included for convenience.
+
+.. automethod:: Mastodon.stream_user
+.. automethod:: Mastodon.stream_public
+.. automethod:: Mastodon.stream_local
+.. automethod:: Mastodon.stream_hashtag
+
+StreamListener
+~~~~~~~~~~~~~~
+
+.. autoclass:: StreamListener
+.. automethod:: StreamListener.on_update
+.. automethod:: StreamListener.on_notification
+.. automethod:: StreamListener.on_delete
+.. automethod:: StreamListener.handle_heartbeat
+
+CallbackStreamListener
+~~~~~~~~~~~~~~~~~~~~~~
+
+.. autoclass:: CallbackStreamListener
 
 .. _Mastodon: https://github.com/tootsuite/mastodon
 .. _Mastodon flagship instance: http://mastodon.social/
diff --git a/mastodon/Mastodon.py b/mastodon/Mastodon.py
index 079fc85..91e6b11 100644
--- a/mastodon/Mastodon.py
+++ b/mastodon/Mastodon.py
@@ -17,10 +17,12 @@ import re
 import copy
 import threading
 import sys
+import six
+
 try:
     from urllib.parse import urlparse
 except ImportError:
-     from urlparse import urlparse
+    from urlparse import urlparse
 
 
 class Mastodon:
@@ -244,6 +246,12 @@ class Mastodon:
 
         Returns a list of toot dicts.
         """
+        if max_id != None:
+            max_id = self.__unpack_id(max_id)
+        
+        if since_id != None:
+            since_id = self.__unpack_id(since_id)
+            
         params_initial = locals()
 
         if timeline == "local":
@@ -289,6 +297,12 @@ class Mastodon:
 
         Returns a list of toot dicts.
         """
+        if max_id != None:
+            max_id = self.__unpack_id(max_id)
+        
+        if since_id != None:
+            since_id = self.__unpack_id(since_id)
+            
         params_initial = locals()        
         
         if local == False:
@@ -308,6 +322,7 @@ class Mastodon:
 
         Returns a toot dict.
         """
+        id = self.__unpack_id(id)
         url = '/api/v1/statuses/{0}'.format(str(id))
         return self.__api_request('GET', url)
 
@@ -318,6 +333,7 @@ class Mastodon:
 
         Returns a card dict.
         """
+        id = self.__unpack_id(id)
         url = '/api/v1/statuses/{0}/card'.format(str(id))
         return self.__api_request('GET', url)
 
@@ -327,6 +343,7 @@ class Mastodon:
 
         Returns a context dict.
         """
+        id = self.__unpack_id(id)
         url = '/api/v1/statuses/{0}/context'.format(str(id))
         return self.__api_request('GET', url)
 
@@ -336,6 +353,7 @@ class Mastodon:
 
         Returns a list of user dicts.
         """
+        id = self.__unpack_id(id)
         url = '/api/v1/statuses/{0}/reblogged_by'.format(str(id))
         return self.__api_request('GET', url)
 
@@ -345,6 +363,7 @@ class Mastodon:
 
         Returns a list of user dicts.
         """
+        id = self.__unpack_id(id)
         url = '/api/v1/statuses/{0}/favourited_by'.format(str(id))
         return self.__api_request('GET', url)
 
@@ -360,10 +379,17 @@ class Mastodon:
 
         Returns a list of notification dicts.
         """
+        if max_id != None:
+            max_id = self.__unpack_id(max_id)
+        
+        if since_id != None:
+            since_id = self.__unpack_id(since_id)
+        
         if id is None:
             params = self.__generate_params(locals(), ['id'])
             return self.__api_request('GET', '/api/v1/notifications', params)
         else:
+            id = self.__unpack_id(id)
             url = '/api/v1/notifications/{0}'.format(str(id))
             return self.__api_request('GET', url)
 
@@ -376,6 +402,7 @@ class Mastodon:
 
         Returns a user dict.
         """
+        id = self.__unpack_id(id)
         url = '/api/v1/accounts/{0}'.format(str(id))
         return self.__api_request('GET', url)
 
@@ -393,6 +420,13 @@ class Mastodon:
 
         Returns a list of toot dicts.
         """
+        id = self.__unpack_id(id)
+        if max_id != None:
+            max_id = self.__unpack_id(max_id)
+        
+        if since_id != None:
+            since_id = self.__unpack_id(since_id)
+            
         params = self.__generate_params(locals(), ['id'])
         url = '/api/v1/accounts/{0}/statuses'.format(str(id))
         return self.__api_request('GET', url, params)
@@ -403,6 +437,13 @@ class Mastodon:
 
         Returns a list of user dicts.
         """
+        id = self.__unpack_id(id)
+        if max_id != None:
+            max_id = self.__unpack_id(max_id)
+        
+        if since_id != None:
+            since_id = self.__unpack_id(since_id)
+            
         params = self.__generate_params(locals(), ['id'])
         url = '/api/v1/accounts/{0}/following'.format(str(id))
         return self.__api_request('GET', url, params)
@@ -413,6 +454,13 @@ class Mastodon:
 
         Returns a list of user dicts.
         """
+        id = self.__unpack_id(id)
+        if max_id != None:
+            max_id = self.__unpack_id(max_id)
+        
+        if since_id != None:
+            since_id = self.__unpack_id(since_id)
+            
         params = self.__generate_params(locals(), ['id'])
         url = '/api/v1/accounts/{0}/followers'.format(str(id))
         return self.__api_request('GET', url, params)
@@ -424,6 +472,7 @@ class Mastodon:
 
         Returns a list of relationship dicts.
         """
+        id = self.__unpack_id(id)
         params = self.__generate_params(locals())
         return self.__api_request('GET', '/api/v1/accounts/relationships',
                                   params)
@@ -460,6 +509,12 @@ class Mastodon:
 
         Returns a list of user dicts.
         """
+        if max_id != None:
+            max_id = self.__unpack_id(max_id)
+        
+        if since_id != None:
+            since_id = self.__unpack_id(since_id)
+            
         params = self.__generate_params(locals())
         return self.__api_request('GET', '/api/v1/mutes', params)
 
@@ -469,6 +524,12 @@ class Mastodon:
 
         Returns a list of user dicts.
         """
+        if max_id != None:
+            max_id = self.__unpack_id(max_id)
+        
+        if since_id != None:
+            since_id = self.__unpack_id(since_id)
+            
         params = self.__generate_params(locals())
         return self.__api_request('GET', '/api/v1/blocks', params)
 
@@ -495,6 +556,12 @@ class Mastodon:
 
         Returns a list of toot dicts.
         """
+        if max_id != None:
+            max_id = self.__unpack_id(max_id)
+        
+        if since_id != None:
+            since_id = self.__unpack_id(since_id)
+            
         params = self.__generate_params(locals())
         return self.__api_request('GET', '/api/v1/favourites', params)
 
@@ -507,6 +574,12 @@ class Mastodon:
 
         Returns a list of user dicts.
         """
+        if max_id != None:
+            max_id = self.__unpack_id(max_id)
+        
+        if since_id != None:
+            since_id = self.__unpack_id(since_id)
+            
         params = self.__generate_params(locals())
         return self.__api_request('GET', '/api/v1/follow_requests', params)
 
@@ -519,6 +592,12 @@ class Mastodon:
 
         Returns a list of blocked domain URLs (as strings, without protocol specifier).
         """
+        if max_id != None:
+            max_id = self.__unpack_id(max_id)
+        
+        if since_id != None:
+            since_id = self.__unpack_id(since_id)
+            
         params = self.__generate_params(locals())
         return self.__api_request('GET', '/api/v1/domain_blocks', params)
 
@@ -555,6 +634,9 @@ class Mastodon:
 
         Returns a toot dict with the new status.
         """
+        if in_reply_to_id != None:
+            in_reply_to_id = self.__unpack_id(in_reply_to_id)
+        
         params_initial = locals()
 
         # Validate visibility parameter
@@ -599,6 +681,7 @@ class Mastodon:
 
         Returns an empty dict for good measure.
         """
+        id = self.__unpack_id(id)
         url = '/api/v1/statuses/{0}'.format(str(id))
         return self.__api_request('DELETE', url)
 
@@ -608,6 +691,7 @@ class Mastodon:
 
         Returns a toot dict with a new status that wraps around the reblogged one.
         """
+        id = self.__unpack_id(id)
         url = '/api/v1/statuses/{0}/reblog'.format(str(id))
         return self.__api_request('POST', url)
 
@@ -617,6 +701,7 @@ class Mastodon:
 
         Returns a toot dict with the status that used to be reblogged.
         """
+        id = self.__unpack_id(id)
         url = '/api/v1/statuses/{0}/unreblog'.format(str(id))
         return self.__api_request('POST', url)
 
@@ -626,6 +711,7 @@ class Mastodon:
 
         Returns a toot dict with the favourited status.
         """
+        id = self.__unpack_id(id)
         url = '/api/v1/statuses/{0}/favourite'.format(str(id))
         return self.__api_request('POST', url)
 
@@ -635,6 +721,7 @@ class Mastodon:
 
         Returns a toot dict with the un-favourited status.
         """
+        id = self.__unpack_id(id)
         url = '/api/v1/statuses/{0}/unfavourite'.format(str(id))
         return self.__api_request('POST', url)
 
@@ -644,6 +731,7 @@ class Mastodon:
 
         Returns a toot dict with the now muted status
         """
+        id = self.__unpack_id(id)
         url = '/api/v1/statuses/{0}/mute'.format(str(id))
         return self.__api_request('POST', url)
 
@@ -653,6 +741,7 @@ class Mastodon:
 
         Returns a toot dict with the status that used to be muted.
         """
+        id = self.__unpack_id(id)
         url = '/api/v1/statuses/{0}/unmute'.format(str(id))
         return self.__api_request('POST', url)
 
@@ -670,6 +759,7 @@ class Mastodon:
         """
         Deletes a single notification
         """
+        id = self.__unpack_id(id)
         params = self.__generate_params(locals())
         return self.__api_request('POST', '/api/v1/notifications/dismiss', params)
 
@@ -682,6 +772,7 @@ class Mastodon:
 
         Returns a relationship dict containing the updated relationship to the user.
         """
+        id = self.__unpack_id(id)
         url = '/api/v1/accounts/{0}/follow'.format(str(id))
         return self.__api_request('POST', url)
 
@@ -700,6 +791,7 @@ class Mastodon:
 
         Returns a relationship dict containing the updated relationship to the user.
         """
+        id = self.__unpack_id(id)
         url = '/api/v1/accounts/{0}/unfollow'.format(str(id))
         return self.__api_request('POST', url)
 
@@ -709,6 +801,7 @@ class Mastodon:
 
         Returns a relationship dict containing the updated relationship to the user.
         """
+        id = self.__unpack_id(id)
         url = '/api/v1/accounts/{0}/block'.format(str(id))
         return self.__api_request('POST', url)
 
@@ -718,6 +811,7 @@ class Mastodon:
 
         Returns a relationship dict containing the updated relationship to the user.
         """
+        id = self.__unpack_id(id)
         url = '/api/v1/accounts/{0}/unblock'.format(str(id))
         return self.__api_request('POST', url)
 
@@ -727,6 +821,7 @@ class Mastodon:
 
         Returns a relationship dict containing the updated relationship to the user.
         """
+        id = self.__unpack_id(id)
         url = '/api/v1/accounts/{0}/mute'.format(str(id))
         return self.__api_request('POST', url)
 
@@ -736,6 +831,7 @@ class Mastodon:
 
         Returns a relationship dict containing the updated relationship to the user.
         """
+        id = self.__unpack_id(id)
         url = '/api/v1/accounts/{0}/unmute'.format(str(id))
         return self.__api_request('POST', url)
 
@@ -763,6 +859,8 @@ class Mastodon:
 
         Returns a report dict.
         """
+        account_id = self.__unpack_id(account_id)
+        status_ids = map(lambda x: self.__unpack_id(x), status_ids)
         params = self.__generate_params(locals())
         return self.__api_request('POST', '/api/v1/reports/', params)
 
@@ -775,6 +873,7 @@ class Mastodon:
 
         Returns an empty dict.
         """
+        id = self.__unpack_id(id)
         url = '/api/v1/follow_requests/{0}/authorize'.format(str(id))
         return self.__api_request('POST', url)
 
@@ -784,6 +883,7 @@ class Mastodon:
 
         Returns an empty dict.
         """
+        id = self.__unpack_id(id)
         url = '/api/v1/follow_requests/{0}/reject'.format(str(id))
         return self.__api_request('POST', url)
 
@@ -911,57 +1011,34 @@ class Mastodon:
     ###
     # Streaming
     ###
-    def user_stream(self, listener, async=False):
+    def stream_user(self, listener, async=False):
         """
         Streams events that are relevant to the authorized user, i.e. home
         timeline and notifications. 'listener' should be a subclass of
         StreamListener which will receive callbacks for incoming events.
-
-        If async is False, this method blocks forever.
-
-        If async is True, 'listener' will listen on another thread and this method
-        will return a handle corresponding to the open connection. The
-        connection may be closed at any time by calling its close() method.
         """
         return self.__stream('/api/v1/streaming/user', listener, async=async)
 
-    def public_stream(self, listener, async=False):
+    def stream_public(self, listener, async=False):
         """
         Streams public events. 'listener' should be a subclass of StreamListener
         which will receive callbacks for incoming events.
-
-        If async is False, this method blocks forever.
-
-        If async is True, 'listener' will listen on another thread and this method
-        will return a handle corresponding to the open connection. The
-        connection may be closed at any time by calling its close() method.
         """
         return self.__stream('/api/v1/streaming/public', listener, async=async)
 
-    def local_stream(self, listener, async=False):
+    def stream_local(self, listener, async=False):
         """
         Streams local events. 'listener' should be a subclass of StreamListener
         which will receive callbacks for incoming events.
 
-        If async is False, this method blocks forever.
-
-        If async is True, 'listener' will listen on another thread and this method
-        will return a handle corresponding to the open connection. The
-        connection may be closed at any time by calling its close() method.
         """
         return self.__stream('/api/v1/streaming/public/local', listener, async=async)
 
-    def hashtag_stream(self, tag, listener, async=False):
+    def stream_hashtag(self, tag, listener, async=False):
         """
         Returns all public statuses for the hashtag 'tag'. 'listener' should be
         a subclass of StreamListener which will receive callbacks for incoming
         events.
-
-        If async is False, this method blocks forever.
-
-        If async is True, 'listener' will listen on another thread and this method
-        will return a handle corresponding to the open connection. The
-        connection may be closed at any time by calling its close() method.
         """
         return self.__stream("/api/v1/streaming/hashtag?tag={}".format(tag), listener)
 
@@ -985,8 +1062,8 @@ class Mastodon:
 
         return (date_time_utc - epoch_utc).total_seconds()
 
-
-    def __json_date_parse(self, json_object):
+    @staticmethod
+    def __json_date_parse(json_object):
         """
         Parse dates in certain known json fields, if possible.
         """
@@ -1002,27 +1079,25 @@ class Mastodon:
                     raise MastodonAPIError('Encountered invalid date.')
         return json_object
 
-    def __json_id_to_bignum(self, json_object):
+    @staticmethod
+    def __json_id_to_bignum(json_object):
         """
         Converts json string IDs to native python bignums.
         """
-        if sys.version_info.major >= 3:
-            str_type = str
-        else:
-            str_type = unicode
-
-        if ('id' in json_object and
-                isinstance(json_object['id'], str_type)):
-            try:
-                json_object['id'] = int(json_object['id'])
-            except ValueError:
-                pass
+        for key in ('id', 'in_reply_to_id', 'in_reply_to_account_id'):
+            if (key in json_object and
+                    isinstance(json_object[key], six.text_type)):
+                try:
+                    json_object[key] = int(json_object[key])
+                except ValueError:
+                    pass
 
         return json_object
-
-    def __json_hooks(self, json_object):
-        json_object = self.__json_date_parse(json_object)
-        json_object = self.__json_id_to_bignum(json_object)
+    
+    @staticmethod
+    def __json_hooks(json_object):
+        json_object = Mastodon.__json_date_parse(json_object)
+        json_object = Mastodon.__json_id_to_bignum(json_object)
         return json_object
 
     def __api_request(self, method, endpoint, params={}, files={}, do_ratelimiting=True):
@@ -1201,9 +1276,16 @@ class Mastodon:
         instance = self.instance()
         if "streaming_api" in instance["urls"] and instance["urls"]["streaming_api"] != self.api_base_url:
             # This is probably a websockets URL, which is really for the browser, but requests can't handle it
-            # So we do this below to turn it into an HTTPS URL
+            # So we do this below to turn it into an HTTPS or HTTP URL
             parse = urlparse(instance["urls"]["streaming_api"])
-            url = "https://" + parse.netloc
+            if parse.scheme == 'wss':
+                url = "https://" + parse.netloc
+            elif parse.scheme == 'ws':
+                url = "http://" + parse.netloc
+            else:
+                raise MastodonAPIError(
+                        "Could not parse streaming api location returned from server: {}.".format(
+                            instance["urls"]["streaming_api"]))
         else:
             url = self.api_base_url
 
@@ -1224,7 +1306,11 @@ class Mastodon:
             def close(self):
                 self.connection.close()
 
+            def is_alive(self):
+                return self._thread.is_alive()
+
             def _threadproc(self):
+                self._thread = threading.current_thread()
                 with closing(connection) as r:
                     try:
                         listener.handle_stream(r.iter_lines())
@@ -1268,7 +1354,20 @@ class Mastodon:
                 del params[key]
 
         return params
-
+    
+    def __unpack_id(self, id):
+        """
+        Internal object-to-id converter
+        
+        Checks if id is a dict that contains id and
+        returns the id inside, otherwise just returns
+        the id straight.
+        """
+        if isinstance(id, dict) and "id" in id:
+            return id["id"]
+        else:
+            return id
+        
     def __get_token_expired(self):
         """Internal helper for oauth code"""
         return self._token_expired < datetime.datetime.now()
@@ -1306,6 +1405,7 @@ class MastodonError(Exception):
 
 
 class MastodonIllegalArgumentError(ValueError, MastodonError):
+    """Raised when an incorrect parameter is passed to a function"""
     pass
 
 
@@ -1314,16 +1414,24 @@ class MastodonIOError(IOError, MastodonError):
 
 
 class MastodonFileNotFoundError(MastodonIOError):
+    """Raised when a file requested to be loaded can not be opened"""
     pass
 
 
 class MastodonNetworkError(MastodonIOError):
+    """Raised when network communication with the server fails"""
     pass
 
 
 class MastodonAPIError(MastodonError):
+    """Raised when the mastodon API generates a response that cannot be handled"""
     pass
 
 
 class MastodonRatelimitError(MastodonError):
+    """Raised when rate limiting is set to manual mode and the rate limit is exceeded"""
+    pass
+
+class MastodonMalformedEventError(MastodonError):
+    """Raised when the server-sent event stream is malformed"""
     pass
diff --git a/mastodon/__init__.py b/mastodon/__init__.py
index 9c8e39b..fdf776d 100644
--- a/mastodon/__init__.py
+++ b/mastodon/__init__.py
@@ -1,4 +1,4 @@
 from mastodon.Mastodon import Mastodon
-from mastodon.streaming import StreamListener, MalformedEventError
+from mastodon.streaming import StreamListener, CallbackStreamListener
 
-__all__ = ['Mastodon', 'StreamListener', 'MalformedEventError']
+__all__ = ['Mastodon', 'StreamListener', 'CallbackStreamListener']
diff --git a/mastodon/streaming.py b/mastodon/streaming.py
index 290ed44..92a02dc 100644
--- a/mastodon/streaming.py
+++ b/mastodon/streaming.py
@@ -4,17 +4,9 @@ https://github.com/tootsuite/mastodon/blob/master/docs/Using-the-API/Streaming-A
 """
 
 import json
-import logging
 import six
-
-
-log = logging.getLogger(__name__)
-
-
-class MalformedEventError(Exception):
-    """Raised when the server-sent event stream is malformed."""
-    pass
-
+from mastodon import Mastodon
+from mastodon.Mastodon import MastodonMalformedEventError
 
 class StreamListener(object):
     """Callbacks for the streaming API. Create a subclass, override the on_xxx
@@ -24,7 +16,7 @@ class StreamListener(object):
 
     def on_update(self, status):
         """A new status has appeared! 'status' is the parsed JSON dictionary
-describing the status."""
+        describing the status."""
         pass
 
     def on_notification(self, notification):
@@ -40,7 +32,8 @@ describing the status."""
         """The server has sent us a keep-alive message. This callback may be
         useful to carry out periodic housekeeping tasks, or just to confirm
         that the connection is still open."""
-
+        pass
+    
     def handle_stream(self, lines):
         """
         Handles a stream of events from the Mastodon server. When each event
@@ -55,7 +48,7 @@ describing the status."""
                 line = raw_line.decode('utf-8')
             except UnicodeDecodeError as err:
                 six.raise_from(
-                    MalformedEventError("Malformed UTF-8", line),
+                    MastodonMalformedEventError("Malformed UTF-8", line),
                     err
                 )
 
@@ -63,7 +56,7 @@ describing the status."""
                 self.handle_heartbeat()
             elif line == '':
                 # end of event
-                self._despatch(event)
+                self._dispatch(event)
                 event = {}
             else:
                 key, value = line.split(': ', 1)
@@ -74,33 +67,65 @@ describing the status."""
                 else:
                     event[key] = value
 
-        # end of stream
-        if event:
-            log.warn("outstanding partial event at end of stream: %s", event)
-
-    def _despatch(self, event):
+    def _dispatch(self, event):
         try:
             name = event['event']
             data = event['data']
-            payload = json.loads(data)
+            payload = json.loads(data, object_hook = Mastodon._Mastodon__json_hooks)
         except KeyError as err:
-            six.raise_from(
-                MalformedEventError('Missing field', err.args[0], event),
-                err
-            )
+           six.raise_from(
+               MastodonMalformedEventError('Missing field', err.args[0], event),
+               err
+           )
         except ValueError as err:
-            # py2: plain ValueError
-            # py3: json.JSONDecodeError, a subclass of ValueError
-            six.raise_from(
-                MalformedEventError('Bad JSON', data),
-                err
-            )
-
+           # py2: plain ValueError
+           # py3: json.JSONDecodeError, a subclass of ValueError
+           six.raise_from(
+               MastodonMalformedEventError('Bad JSON', data),
+               err
+           )
+           
         handler_name = 'on_' + name
         try:
             handler = getattr(self, handler_name)
-        except AttributeError:
-            log.warn("Unhandled event '%s'", name)
+        except AttributeError as err:
+            six.raise_from(
+               MastodonMalformedEventError('Bad event type', name),
+               err
+            )
         else:
             # TODO: allow handlers to return/raise to stop streaming cleanly
             handler(payload)
+
+class CallbackStreamListener(StreamListener):
+    """
+    Simple callback stream handler class.
+    Can optionally additionally send local update events to a separate handler.
+    """
+    def __init__(self, update_handler = None, local_update_handler = None, delete_handler = None, notification_handler = None):
+        super(CallbackStreamListener, self).__init__()
+        self.update_handler = update_handler
+        self.local_update_handler = local_update_handler
+        self.delete_handler = delete_handler
+        self.notification_handler = notification_handler
+        
+    def on_update(self, status):
+        if self.update_handler != None:
+            self.update_handler(status)
+        
+        try:
+            if self.local_update_handler != None and not "@" in status["account"]["acct"]:
+                self.local_update_handler(status)
+        except Exception as err:
+            six.raise_from(
+               MastodonMalformedEventError('received bad update', status),
+               err
+            )
+    
+    def on_delete(self, deleted_id):
+        if self.delete_handler != None:
+            self.delete_handler(deleted_id)
+    
+    def on_notification(self, notification):
+        if self.notification_handler != None:
+            self.notification_handler(notification)
\ No newline at end of file
diff --git a/tests/test_streaming.py b/tests/test_streaming.py
index 1f8d75f..33a3f9a 100644
--- a/tests/test_streaming.py
+++ b/tests/test_streaming.py
@@ -1,7 +1,9 @@
 import six
 import pytest
 import itertools
-from mastodon.streaming import StreamListener, MalformedEventError
+from mastodon.streaming import StreamListener
+from mastodon.Mastodon import MastodonMalformedEventError
+
 
 
 class Listener(StreamListener):
@@ -20,6 +22,9 @@ class Listener(StreamListener):
     def on_delete(self, status_id):
         self.deletes.append(status_id)
 
+    def on_blahblah(self, data):
+        pass
+
     def handle_heartbeat(self):
         self.heartbeats += 1
 
@@ -101,7 +106,7 @@ def test_unknown_event():
 
 def test_missing_event_name():
     listener = Listener()
-    with pytest.raises(MalformedEventError):
+    with pytest.raises(MastodonMalformedEventError):
         listener.handle_stream_([
             'data: {}',
             '',
@@ -115,7 +120,7 @@ def test_missing_event_name():
 
 def test_missing_data():
     listener = Listener()
-    with pytest.raises(MalformedEventError):
+    with pytest.raises(MastodonMalformedEventError):
         listener.handle_stream_([
             'event: update',
             '',
@@ -169,7 +174,7 @@ def test_valid_utf8():
 def test_invalid_utf8():
     """Cat Face With Tears O"""
     listener = Listener()
-    with pytest.raises(MalformedEventError):
+    with pytest.raises(MastodonMalformedEventError):
         listener.handle_stream_([
             'event: update',
             'data: {"foo": "\xF0\x9F\x98"}',