Repository: pystorm/pystorm

Issue:
Support Spout activate/deactivate
_From @andrewkcarter on January 23, 2017 21:51_

Storm has recently exposed the activate and deactivate events via the `ShellSpout`. I would like to see these made available in streamparse for storm's next release.

https://github.com/apache/storm/pull/1096

_Copied from original issue: Parsely/streamparse#351_

Discussion:
coveralls: 
[![Coverage Status](https://coveralls.io/builds/9818486/badge)](https://coveralls.io/builds/9818486)

Coverage decreased (-0.1%) to 83.091% when pulling **9d1e2b94ba2b130cd99f0993bf7a4fc37bf1ae19 on feature/activate_deactivate_spout** into **266004865f5c9db2b8fb96e1b4c0217f19aefc9a on master**.

coveralls: 
[![Coverage Status](https://coveralls.io/builds/9818530/badge)](https://coveralls.io/builds/9818530)

Coverage decreased (-0.1%) to 83.091% when pulling **552fe17df9b072ac8721e4bad135651165afd0e5 on feature/activate_deactivate_spout** into **266004865f5c9db2b8fb96e1b4c0217f19aefc9a on master**.


Patch:
diff --git a/pystorm/spout.py b/pystorm/spout.py
index 47cfd37..169fbd1 100644
--- a/pystorm/spout.py
+++ b/pystorm/spout.py
@@ -76,6 +76,23 @@ class Spout(Component):
                                        direct_task=direct_task,
                                        need_task_ids=need_task_ids)
 
+    def activate(self):
+        """Called when the Spout has been activated after being deactivated.
+
+        .. note::
+            This requires at least Storm 1.1.0.
+        """
+        pass
+
+
+    def deactivate(self):
+        """Called when the Spout has been deactivated.
+
+        .. note::
+            This requires at least Storm 1.1.0.
+        """
+        pass
+
     def _run(self):
         """The inside of ``run``'s infinite loop.
 
@@ -88,6 +105,10 @@ class Spout(Component):
             self.ack(cmd['id'])
         elif cmd['command'] == 'fail':
             self.fail(cmd['id'])
+        elif cmd['command'] == 'activate':
+            self.activate()
+        elif cmd['command'] == 'deactivate':
+            self.deactivate()
         else:
             self.logger.error('Received invalid command from Storm: %r', cmd)
         self.send_message({'command': 'sync'})


Test patch:
diff --git a/test/pystorm/test_spout.py b/test/pystorm/test_spout.py
index ed84b16..25c762a 100644
--- a/test/pystorm/test_spout.py
+++ b/test/pystorm/test_spout.py
@@ -87,6 +87,22 @@ class SpoutTests(unittest.TestCase):
         read_command_mock.assert_called_with(self.spout)
         self.assertEqual(next_tuple_mock.call_count, 1)
 
+    @patch.object(Spout, 'read_command', autospec=True,
+                  return_value={'command': 'activate', 'id': 1234})
+    @patch.object(Spout, 'activate', autospec=True)
+    def test_activate(self, activate_mock, read_command_mock):
+        self.spout._run()
+        read_command_mock.assert_called_with(self.spout)
+        self.assertEqual(activate_mock.call_count, 1)
+
+    @patch.object(Spout, 'read_command', autospec=True,
+                  return_value={'command': 'deactivate', 'id': 1234})
+    @patch.object(Spout, 'deactivate', autospec=True)
+    def test_deactivate(self, deactivate_mock, read_command_mock):
+        self.spout._run()
+        read_command_mock.assert_called_with(self.spout)
+        self.assertEqual(deactivate_mock.call_count, 1)
+
 
 class ReliableSpoutTests(unittest.TestCase):