3 files changed, 66 insertions, 21 deletions

diff --git a/docs/index.rst b/docs/index.rst
index 4d7073c..fc8c003 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -546,6 +546,26 @@ Push subscription dicts
                   # if webpushes have been requested for those events.
     }
     
+Push notification dicts
+~~~~~~~~~~~~~~~~~~~~~~~
+.. _push notification dict:
+
+.. code-block:: python
+
+    mastodon.push_subscription_decrypt_push(...)
+    # Returns the following dictionary
+    {
+        'access_token': # Access token that can be used to access the API as the 
+                        # notified user
+        'body': # Text body of the notification
+        'icon': # URL to an icon for the notification
+        'notification_id': # ID that can be passed to notification() to get the full 
+                           # notification object,
+        'notification_type': # 'mention', 'reblog', 'follow' or 'favourite'
+        'preferred_locale': # The users preferred locale
+        'title': # Title for the notification
+    }
+    
 App registration and user authentication
 ----------------------------------------
 Before you can use the mastodon API, you have to register your 
@@ -795,7 +815,8 @@ will return a handle corresponding to the open connection. If, in addition, `asy
 the thread will attempt to reconnect to the streaming API if any errors are encountered, waiting
 `async_reconnect_wait_sec` seconds between reconnection attempts. Note that no effort is made
 to "catch up" - events created while the connection is broken will not be received. If you need to make
-sure to get absolutely all notifications / deletes / toots, you will have to do that manually.
+sure to get absolutely all notifications / deletes / toots, you will have to do that manually, e.g.
+using the `on_abort` handler to fill in events since the last received one and then reconnecting.
 
 The connection may be closed at any time by calling the handles close() method. The 
 current status of the handler thread can be checked with the handles is_alive() function,
diff --git a/mastodon/Mastodon.py b/mastodon/Mastodon.py
index c2da613..b5c82f8 100644
--- a/mastodon/Mastodon.py
+++ b/mastodon/Mastodon.py
@@ -132,6 +132,7 @@ class Mastodon:
     __DICT_VERSION_ACTIVITY = "2.1.2" 
     __DICT_VERSION_REPORT = "1.1.0"
     __DICT_VERSION_PUSH = "2.4.0"
+    __DICT_VERSION_PUSH_NOTIF = "2.4.0"
     
     ###
     # Registering apps
@@ -1609,13 +1610,14 @@ class Mastodon:
         
         return priv_dict, pub_dict
     
+    @api_version("2.4.0", "2.4.0", __DICT_VERSION_PUSH_NOTIF)
     def push_subscription_decrypt_push(self, data, decrypt_params, encryption_header, crypto_key_header):
         """
         Decrypts `data` received in a webpush request. Requires the private key dict 
         from `push_subscription_generate_keys()`_ (`decrypt_params`) as well as the 
         Encryption and server Crypto-Key headers from the received webpush
         
-        Returns the decoded webpush.
+        Returns the decoded webpush as a `push notification dict`_.
         """
         salt = self.__decode_webpush_b64(encryption_header.split("salt=")[1].strip())
         dhparams = self.__decode_webpush_b64(crypto_key_header.split("dh=")[1].split(";")[0].strip())
diff --git a/mastodon/streaming.py b/mastodon/streaming.py
index 8c3ec19..65ec30a 100644
--- a/mastodon/streaming.py
+++ b/mastodon/streaming.py
@@ -25,8 +25,15 @@ class StreamListener(object):
         describing the notification."""
         pass
 
-    def on_abort(self):
-        """There was a connection error or read timeout."""
+    def on_abort(self, err):
+        """There was a connection error, read timeout or other error fatal to
+        the streaming connection. The exception object about to be raised
+        is passed to this function for reference.
+        
+        Note that the exception will be raised properly once you return from this
+        function, so if you are using this handler to reconnect, either never
+        return or start a thread and then catch and ignore the exception.
+        """
         pass
 
     def on_delete(self, status_id):
@@ -55,8 +62,10 @@ class StreamListener(object):
                         try:
                             line = line_buffer.decode('utf-8')
                         except UnicodeDecodeError as err:
+                            exception = MastodonMalformedEventError("Malformed UTF-8")
+                            self.on_abort(exception)
                             six.raise_from(
-                                MastodonMalformedEventError("Malformed UTF-8"),
+                                exception,
                                 err
                             )
                         if line == '':
@@ -68,15 +77,17 @@ class StreamListener(object):
                     else:
                         line_buffer.extend(chunk)
         except ChunkedEncodingError as err:
-            self.on_abort()
+            exception = MastodonNetworkError("Server ceased communication.")
+            self.on_abort(exception)
             six.raise_from(
-                MastodonNetworkError("Server ceased communication."),
+                exception,
                 err
             )
         except MastodonReadTimeout as err:
-            self.on_abort()
+            exception = MastodonReadTimeout("Timed out while reading from server."),
+            self.on_abort(exception)
             six.raise_from(
-                MastodonReadTimeout("Timed out while reading from server."),
+                exception,
                 err
             )
 
@@ -84,7 +95,12 @@ class StreamListener(object):
         if line.startswith(':'):
             self.handle_heartbeat()
         else:
-            key, value = line.split(': ', 1)
+            try:
+                key, value = line.split(': ', 1)
+            except:
+                exception = MastodonMalformedEventError("Malformed event.")
+                self.on_abort(exception)
+                raise exception
             # According to the MDN spec, repeating the 'data' key
             # represents a newline(!)
             if key in event:
@@ -99,24 +115,30 @@ class StreamListener(object):
             data = event['data']
             payload = json.loads(data, object_hook = Mastodon._Mastodon__json_hooks)
         except KeyError as err:
-           six.raise_from(
-               MastodonMalformedEventError('Missing field', err.args[0], event),
-               err
-           )
+            exception = MastodonMalformedEventError('Missing field', err.args[0], event)
+            self.on_abort(exception)
+            six.raise_from(
+                exception,
+                err
+            )
         except ValueError as err:
-           # py2: plain ValueError
-           # py3: json.JSONDecodeError, a subclass of ValueError
-           six.raise_from(
-               MastodonMalformedEventError('Bad JSON', data),
-               err
-           )
+            # py2: plain ValueError
+            # py3: json.JSONDecodeError, a subclass of ValueError
+            exception = MastodonMalformedEventError('Bad JSON', data)
+            self.on_abort(exception)
+            six.raise_from(
+                exception,
+                err
+            )
            
         handler_name = 'on_' + name
         try:
             handler = getattr(self, handler_name)
         except AttributeError as err:
+            exception = MastodonMalformedEventError('Bad event type', name)
+            self.on_abort(exception)
             six.raise_from(
-               MastodonMalformedEventError('Bad event type', name),
+               exception,
                err
             )
         else: