instantiate dispatcher once per job run

* Instantiating the callback dispatch queue on each job event callback
is expensive. Instead, instantiate it only once. Note, we do not need to
instantiate the callback queue in the iso case so we do not.

awx/main/tasks.py

@@ -1026,9 +1026,8 @@ class BaseTask(object):
             if event_data[self.event_data_key] != 'job_id':
                 event_data.pop('parent_uuid', None)
         should_write_event = False
-        dispatcher = CallbackQueueDispatcher()
         event_data.setdefault(self.event_data_key, self.instance.id)
-        dispatcher.dispatch(event_data)
+        self.dispatcher.dispatch(event_data)
         self.event_ct += 1
 
         '''
@@ -1056,13 +1055,12 @@ class BaseTask(object):
         '''
         Ansible runner callback triggered on finished run
         '''
-        dispatcher = CallbackQueueDispatcher()
         event_data = {
             'event': 'EOF',
             'final_counter': self.event_ct,
         }
         event_data.setdefault(self.event_data_key, self.instance.id)
-        dispatcher.dispatch(event_data)
+        self.dispatcher.dispatch(event_data)
 
     def status_handler(self, status_data, runner_config):
         '''
@@ -1244,6 +1242,7 @@ class BaseTask(object):
                                                            ident=str(self.instance.pk))
                 self.event_ct = len(isolated_manager_instance.handled_events)
             else:
+                self.dispatcher = CallbackQueueDispatcher()
                 res = ansible_runner.interface.run(**params)
                 status = res.status
                 rc = res.rc

awx/main/tests/unit/test_tasks.py

@@ -87,13 +87,6 @@ def adhoc_update_model_wrapper(adhoc_job):
     return fn
 
 
-@pytest.fixture
-def patch_CallbackQueueDispatcher():
-    with mock.patch('awx.main.tasks.CallbackQueueDispatcher') as m:
-        m.return_value = m
-        yield m
-
-
 def test_send_notifications_not_list():
     with pytest.raises(TypeError):
         tasks.send_notifications(None)
@@ -399,8 +392,9 @@ class TestGenericRun():
         ]:
             assert c in task.update_model.call_args_list
 
-    def test_event_count(self, patch_CallbackQueueDispatcher):
+    def test_event_count(self):
         task = tasks.RunJob()
+        task.dispatcher = mock.MagicMock()
         task.instance = Job()
         task.event_ct = 0
         event_data = {}
@@ -408,12 +402,13 @@ class TestGenericRun():
         [task.event_handler(event_data) for i in range(20)]
         assert 20 == task.event_ct
 
-    def test_finished_callback_eof(self, patch_CallbackQueueDispatcher):
+    def test_finished_callback_eof(self):
         task = tasks.RunJob()
+        task.dispatcher = mock.MagicMock()
         task.instance = Job(pk=1, id=1)
         task.event_ct = 17
         task.finished_callback(None)
-        patch_CallbackQueueDispatcher.dispatch.assert_called_with({'event': 'EOF', 'final_counter': 17, 'job_id': 1})
+        task.dispatcher.dispatch.assert_called_with({'event': 'EOF', 'final_counter': 17, 'job_id': 1})
 
     def test_save_job_metadata(self, job, update_model_wrapper):
         class MockMe():