Merge pull request #2046 from coolo/ack_pika

Fix rabbitmq acknowledge: first ACK then check the content

PubSubConsumer.py

@@ -271,7 +271,7 @@ class PubSubConsumer(object):
         :param int delivery_tag: The delivery tag from the Basic.Deliver frame
 
         """
-        self.logger.info('Acknowledging message %s', delivery_tag)
+        self.logger.debug('Acknowledging message %s', delivery_tag)
         self._channel.basic_ack(delivery_tag)
 
     def on_cancelok(self, _unused_frame, userdata):

gocd/rabbit-openqa.py

@@ -213,13 +213,13 @@ class Listener(PubSubConsumer):
             p.openqa_job_change(iso)
 
     def on_message(self, unused_channel, method, properties, body):
+        self.acknowledge_message(method.delivery_tag)
         if method.routing_key == '{}.obs.repo.published'.format(amqp_prefix):
             self.on_published_repo(json.loads(body))
         elif re.search(r'.openqa.', method.routing_key):
             self.on_openqa_job(json.loads(body).get('ISO'))
         else:
             self.logger.warning("unknown rabbitmq message {}".format(method.routing_key))
-        self.acknowledge_message(method.delivery_tag)
 
 if __name__ == '__main__':
     parser = argparse.ArgumentParser(

gocd/rabbit-repoid.py

@@ -86,6 +86,7 @@ class Listener(PubSubConsumer):
                 f.write('{}: {}\n'.format(arch, ids[arch]))
 
     def on_message(self, unused_channel, method, properties, body):
+        self.acknowledge_message(method.delivery_tag)
         try:
             body = json.loads(body)
         except ValueError:
@@ -100,7 +101,6 @@ class Listener(PubSubConsumer):
         else:
             self.logger.warning(
                 'unknown rabbitmq message {}'.format(method.routing_key))
-        self.acknowledge_message(method.delivery_tag)
 
 if __name__ == '__main__':
     parser = argparse.ArgumentParser(