Change zmq socket characterstics to ack response. Fix up unit tests

awx/main/management/commands/run_callback_receiver.py

@@ -7,6 +7,7 @@ import datetime
 import logging
 import json
 import signal
+import time
 from optparse import make_option
 from multiprocessing import Process
 
@@ -35,7 +36,7 @@ def run_subscriber(consumer_port, queue_port, use_workers=True):
         return _handler
         
     consumer_context = zmq.Context()
-    consumer_subscriber = consumer_context.socket(zmq.PULL)
+    consumer_subscriber = consumer_context.socket(zmq.REP)
     consumer_subscriber.bind(consumer_port)
 
     queue_context = zmq.Context()
@@ -57,6 +58,7 @@ def run_subscriber(consumer_port, queue_port, use_workers=True):
             queue_publisher.send_json(message)
         else:
             process_job_event(message)
+        consumer_subscriber.send("1")
 
 
 @transaction.commit_on_success
@@ -89,12 +91,12 @@ def process_job_event(data):
             break
         except DatabaseError as e:
             transaction.rollback()
-            logger.debug('Database error saving job event, retrying in '
-                          '1 second (retry #%d): %s', retry_count + 1, e)
+            print('Database error saving job event, retrying in '
+                  '1 second (retry #%d): %s', retry_count + 1, e)
             time.sleep(1)
     else:
-        logger.error('Failed to save job event after %d retries.',
-                     retry_count)
+        print('Failed to save job event after %d retries.',
+              retry_count)
 
 def callback_worker(port):
     pool_context = zmq.Context()

awx/main/tests/jobs.py

@@ -911,9 +911,6 @@ class JobStartCancelTest(BaseJobTestMixin, django.test.LiveServerTestCase):
         job = self.job_ops_east_run
         job.start()
 
-        # Wait for events to filter in since we are using a single consumer
-        time.sleep(30)
-        
         # Check that the job detail has been updated.
         url = reverse('api:job_detail', args=(job.pk,))
         with self.current_user(self.user_sue):

awx/plugins/callback/job_event_callback.py

@@ -37,6 +37,7 @@ import os
 import sys
 import urllib
 import urlparse
+import time
 
 import requests
 
@@ -109,7 +110,7 @@ class CallbackModule(object):
 
     def _start_connection(self):
         self.context = zmq.Context()
-        self.socket = self.context.socket(zmq.PUSH)
+        self.socket = self.context.socket(zmq.REQ)
         self.socket.connect("tcp://127.0.0.1:5556")
 
     def _post_job_event_queue_msg(self, event, event_data):
@@ -134,6 +135,7 @@ class CallbackModule(object):
                     self._start_connection()
 
                 self.socket.send_json(msg)
+                self.socket.recv()
                 return
             except Exception, e:
                 self.logger.info('Publish Exception: %r, retry=%d', e,