1 files changed, 58 insertions, 33 deletions

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