diff --git a/docs/index.rst b/docs/index.rst
index d608520..9cd5c09 100755
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -663,6 +663,15 @@ user-friendly, but nevertheless it should be usable.)
    a separate stream. The callable should not raise any exceptions (unless it wants
    the current operation to fail).
 
+.. versionadded:: 0.5.4
+   Instances of the result classes from operations now have an ``on_data_failure``
+   attribute, which defaults to ``None``. If an ``on_data`` callable raises an exception,
+   the ``on_data_failure`` attribute of the returned object from a high-level operation
+   is set to the first exception that was raised. The ``on_data`` callable will continue
+   to be called with future chunks. If you use ``on_data`` with code that can raise any
+   exceptions, be sure to check the ``on_data_failure`` attribute of a returned object
+   before using any other aspects of the result.
+
 .. versionadded:: 0.4.2
    Information on keys returned by :meth:`~gnupg.GPG.list_keys` or
    :meth:`~gnupg.GPG.scan_keys` now incudes a ``subkey_info`` dictionary, which
diff --git a/gnupg.py b/gnupg.py
index e8971c6..86d46d8 100644
--- a/gnupg.py
+++ b/gnupg.py
@@ -217,6 +217,8 @@ class StatusHandler(object):
     The base class for handling status messages from `gpg`.
     """
 
+    on_data_failure = None  # set at instance level when failures occur
+
     def __init__(self, gpg):
         """
         Initialize an instance.
@@ -1249,17 +1251,27 @@ def _read_data(self, stream, result, on_data=None, buffer_size=1024):
         # Read the contents of the file from GPG's stdout
         assert buffer_size > 0
         chunks = []
+        on_data_failure = None
         while True:
             data = stream.read(buffer_size)
             if len(data) == 0:
                 if on_data:
-                    on_data(data)
+                    try:
+                        on_data(data)
+                    except Exception as e:
+                        if on_data_failure is None:
+                            on_data_failure = e
                 break
             if log_everything:
                 logger.debug('chunk: %r' % data[:256])
             append = True
             if on_data:
-                append = on_data(data) is not False
+                try:
+                    on_data_result = on_data(data)
+                    append = on_data_result is not False
+                except Exception as e:
+                    if on_data_failure is None:
+                        on_data_failure = e
             if append:
                 chunks.append(data)
         if _py3k:
@@ -1267,6 +1279,8 @@ def _read_data(self, stream, result, on_data=None, buffer_size=1024):
             result.data = type(data)().join(chunks)
         else:
             result.data = ''.join(chunks)
+        if on_data_failure:
+            result.on_data_failure = on_data_failure
 
     def _collect_output(self, process, result, writer=None, stdin=None):
         """
diff --git a/test_gnupg.py b/test_gnupg.py
index 2232553..7675a7e 100644
--- a/test_gnupg.py
+++ b/test_gnupg.py
@@ -832,6 +832,7 @@ def collector(data):
         gpg.on_data = collector
         result = gpg.encrypt(data, barbara)
         self.assertEqual(0, result.returncode, 'Non-zero return code')
+        self.assertIsNone(result.on_data_failure)
         edata = str(result)
         self.assertTrue(chunks)
         expected = type(chunks[0])().join(chunks)
@@ -840,9 +841,21 @@ def collector(data):
         ddata = gpg.decrypt(edata, passphrase='bbrown')
         self.assertEqual(0, ddata.returncode, 'Non-zero return code')
         self.assertEqual(data.encode('ascii'), ddata.data, 'Round-trip must work')
+        self.assertIsNone(result.on_data_failure)
         expected = type(chunks[0])().join(chunks)
         self.assertEqual(expected.decode('ascii'), data)
 
+        # test with on-data  generating an exception
+
+        def exceptor(data):
+            raise ValueError('exception in on_data')
+
+        chunks = []
+        gpg.on_data = exceptor
+        ddata = gpg.decrypt(edata, passphrase='bbrown')
+        self.assertIs(type(ddata.on_data_failure), ValueError)
+        self.assertEqual(str(ddata.on_data_failure), 'exception in on_data')
+
         # test signing with encryption and verification during decryption
         logger.debug('encrypting with signature')
         gpg.on_data = None