Repository: pystorm/pystorm

Issue:
IOError is raised instead of StormWentAwayError if pipe breaks during send_message
```
IOError: [Errno 32] Broken pipe
(2 additional frame(s) were not displayed)
...
  File "pystorm/component.py", line 471, in run
    self._handle_run_exception(e)
  File "pystorm/component.py", line 483, in _handle_run_exception
    self.raise_exception(exc)
  File "pystorm/component.py", line 321, in raise_exception
    self.send_message({'command': 'error', 'msg': str(message)})
  File "pystorm/component.py", line 305, in send_message
    self.serializer.send_message(message)
  File "pystorm/serializers/serializer.py", line 26, in send_message
    self.output_stream.flush()
```

This should be a `StormWentAwayError`.  Will fix ASAP.

Discussion:

Patch:
diff --git a/pystorm/serializers/serializer.py b/pystorm/serializers/serializer.py
index 03a15de..58f6a0e 100644
--- a/pystorm/serializers/serializer.py
+++ b/pystorm/serializers/serializer.py
@@ -4,6 +4,8 @@ each serializer a Java counterpart needs to exist.
 
 from __future__ import absolute_import, print_function, unicode_literals
 
+from ..exceptions import StormWentAwayError
+
 
 class Serializer(object):
 
@@ -21,9 +23,12 @@ class Serializer(object):
     def send_message(self, msg_dict):
         """Serialize a message dictionary and write it to the output stream."""
         with self._writer_lock:
-            self.output_stream.flush()
-            self.output_stream.write(self.serialize_dict(msg_dict))
-            self.output_stream.flush()
+            try:
+                self.output_stream.flush()
+                self.output_stream.write(self.serialize_dict(msg_dict))
+                self.output_stream.flush()
+            except IOError:
+                raise StormWentAwayError()
 
     def serialize_dict(self, msg_dict):
         """Convert a message dictionary to bytes.  Used by send_message"""


Test patch:
diff --git a/test/pystorm/serializers/test_json_serializer.py b/test/pystorm/serializers/test_json_serializer.py
index a18a9e8..f460f6e 100644
--- a/test/pystorm/serializers/test_json_serializer.py
+++ b/test/pystorm/serializers/test_json_serializer.py
@@ -7,7 +7,9 @@ except ImportError:
     import mock
 
 import simplejson as json
+import pytest
 
+from pystorm.exceptions import StormWentAwayError
 from pystorm.serializers.json_serializer import JSONSerializer
 
 from .serializer import SerializerTestCase
@@ -33,3 +35,12 @@ class TestJSONSerializer(SerializerTestCase):
         self.instance.output_stream = StringIO()
         self.instance.send_message(msg_dict)
         assert self.instance.output_stream.getvalue() == expected_output
+
+    def test_send_message_raises_stormwentaway(self):
+        string_io_mock = mock.MagicMock(autospec=True)
+        def raiser(): # lambdas can't raise
+            raise IOError()
+        string_io_mock.flush.side_effect = raiser
+        self.instance.output_stream = string_io_mock
+        with pytest.raises(StormWentAwayError):
+            self.instance.send_message({'hello': "world",})
diff --git a/test/pystorm/serializers/test_msgpack_serializer.py b/test/pystorm/serializers/test_msgpack_serializer.py
index e36a1ec..d51c19f 100644
--- a/test/pystorm/serializers/test_msgpack_serializer.py
+++ b/test/pystorm/serializers/test_msgpack_serializer.py
@@ -7,7 +7,9 @@ except ImportError:
     import mock
 
 import msgpack
+import pytest
 
+from pystorm.exceptions import StormWentAwayError
 from pystorm.serializers.msgpack_serializer import MsgpackSerializer
 
 from .serializer import SerializerTestCase
@@ -32,3 +34,12 @@ class TestMsgpackSerializer(SerializerTestCase):
         expected_output = msgpack.packb(msg_dict)
         self.instance.send_message(msg_dict)
         assert self.instance.output_stream.getvalue() == expected_output
+
+    def test_send_message_raises_stormwentaway(self):
+        bytes_io_mock = mock.MagicMock(autospec=True)
+        def raiser(): # lambdas can't raise
+            raise IOError()
+        bytes_io_mock.flush.side_effect = raiser
+        self.instance.output_stream = bytes_io_mock
+        with pytest.raises(StormWentAwayError):
+            self.instance.send_message({'hello': "world",})