FEATURE: handler listenersproposal_handler_pub_sub

Fixes ansible/proposals#8

5 files changed, 28 insertions, 9 deletions

diff --git a/lib/ansible/executor/play_iterator.py b/lib/ansible/executor/play_iterator.py
index 0b46c3925e..c6f0a67049 100644
--- a/lib/ansible/executor/play_iterator.py
+++ b/lib/ansible/executor/play_iterator.py
@@ -212,9 +212,6 @@ class PlayIterator:
             # plays won't try to advance)
             play_context.start_at_task = None
 
-        # Extend the play handlers list to include the handlers defined in roles
-        self._play.handlers.extend(play.compile_roles_handlers())
-
     def get_host_state(self, host):
         try:
             return self._host_states[host.name].copy()
diff --git a/lib/ansible/executor/task_queue_manager.py b/lib/ansible/executor/task_queue_manager.py
index 8373264ef1..ae7190c003 100644
--- a/lib/ansible/executor/task_queue_manager.py
+++ b/lib/ansible/executor/task_queue_manager.py
@@ -86,6 +86,7 @@ class TaskQueueManager:
 
         # this dictionary is used to keep track of notified handlers
         self._notified_handlers = dict()
+        self._listening_handlers = dict()
 
         # dictionaries to keep track of failed/unreachable hosts
         self._failed_hosts      = dict()
@@ -118,6 +119,8 @@ class TaskQueueManager:
         # Proxied dicts don't support iteritems, so we have to use keys()
         for key in self._notified_handlers.keys():
             del self._notified_handlers[key]
+        for key in self._listening_handlers.keys():
+            del self._listening_handlers[key]
 
         def _process_block(b):
             temp_list = []
@@ -135,6 +138,10 @@ class TaskQueueManager:
         # then initialize it with the handler names from the handler list
         for handler in handler_list:
             self._notified_handlers[handler.get_name()] = []
+            if handler.listen:
+                if handler.listen not in self._listening_handlers:
+                    self._listening_handlers[handler.listen] = []
+                self._listening_handlers[handler.listen].append(handler.get_name())
 
     def load_callbacks(self):
         '''
@@ -219,6 +226,7 @@ class TaskQueueManager:
         self.send_callback('v2_playbook_on_play_start', new_play)
 
         # initialize the shared dictionary containing the notified handlers
+        new_play.handlers = new_play.compile_roles_handlers() + new_play.handlers
         self._initialize_notified_handlers(new_play.handlers)
 
         # load the specified strategy (or the default linear one)
diff --git a/lib/ansible/playbook/handler.py b/lib/ansible/playbook/handler.py
index c8c1572e48..a611b72259 100644
--- a/lib/ansible/playbook/handler.py
+++ b/lib/ansible/playbook/handler.py
@@ -20,11 +20,13 @@ from __future__ import (absolute_import, division, print_function)
 __metaclass__ = type
 
 from ansible.errors import AnsibleError
-#from ansible.inventory.host import Host
+from ansible.playbook.attribute import FieldAttribute
 from ansible.playbook.task import Task
 
 class Handler(Task):
 
+    _listen = FieldAttribute(isa='list')
+
     def __init__(self, block=None, role=None, task_include=None):
         self._flagged_hosts = []
 
diff --git a/lib/ansible/plugins/strategy/__init__.py b/lib/ansible/plugins/strategy/__init__.py
index e3bc2ac747..84eafa004f 100644
--- a/lib/ansible/plugins/strategy/__init__.py
+++ b/lib/ansible/plugins/strategy/__init__.py
@@ -101,6 +101,7 @@ class StrategyBase:
         self._inventory         = tqm.get_inventory()
         self._workers           = tqm.get_workers()
         self._notified_handlers = tqm.get_notified_handlers()
+        self._listening_handlers = tqm._listening_handlers
         self._variable_manager  = tqm.get_variable_manager()
         self._loader            = tqm.get_loader()
         self._final_q           = tqm._final_q
@@ -316,12 +317,21 @@ class StrategyBase:
 
                     original_host = get_original_host(task_result._host)
                     original_task = iterator.get_original_task(original_host, task_result._task)
-                    if handler_name not in self._notified_handlers:
-                        self._notified_handlers[handler_name] = []
 
-                    if original_host not in self._notified_handlers[handler_name]:
-                        self._notified_handlers[handler_name].append(original_host)
-                        display.vv("NOTIFIED HANDLER %s" % (handler_name,))
+                    if handler_name not in self._notified_handlers:
+                        print("%s wasn't in _notified_handlers")
+                        if handler_name in self._listening_handlers:
+                            for listening_handler in self._listening_handlers[handler_name]:
+                                if original_host not in self._notified_handlers[listening_handler]:
+                                    self._notified_handlers[listening_handler].append(original_host)
+                                    display.vv("NOTIFIED HANDLER %s" % (listening_handler,))
+                        else:
+                            self._notified_handlers[handler_name] = [original_host]
+                            display.vv("NOTIFIED HANDLER %s" % (handler_name,))
+                    else:
+                        if original_host not in self._notified_handlers[handler_name]:
+                            self._notified_handlers[handler_name].append(original_host)
+                            display.vv("NOTIFIED HANDLER %s" % (handler_name,))
 
                 elif result[0] == 'register_host_var':
                     # essentially the same as 'set_host_var' below, however we
diff --git a/test/units/plugins/strategies/test_strategy_base.py b/test/units/plugins/strategies/test_strategy_base.py
index 9ea944a2a1..8748f2ec4b 100644
--- a/test/units/plugins/strategies/test_strategy_base.py
+++ b/test/units/plugins/strategies/test_strategy_base.py
@@ -45,12 +45,14 @@ class TestStrategyBase(unittest.TestCase):
         mock_tqm = MagicMock(TaskQueueManager)
         mock_tqm._final_q = MagicMock()
         mock_tqm._options = MagicMock()
+        mock_tqm._listening_handlers = []
         strategy_base = StrategyBase(tqm=mock_tqm)
 
     def test_strategy_base_run(self):
         mock_tqm = MagicMock(TaskQueueManager)
         mock_tqm._final_q = MagicMock()
         mock_tqm._stats = MagicMock()
+        mock_tqm._listening_handlers = []
         mock_tqm.send_callback.return_value = None
 
         mock_iterator  = MagicMock()