Merge branch 'feat-redesign-mq' into feat-subscribe-mail-lists

README.md

@@ -41,7 +41,10 @@ def callback_function(ch, method, properties, body):
 # start consume messagre from queue with callback function
 rc_rmq.start_consume({
   'queue': 'queue_name',
+  'routing_key: 'your_key',
   'cb': callback_function
 })
 
+# don't forget to close connection
+rc_rmq.disconnect()
 ```

create_account.py

+#!/usr/bin/env python
+import sys
+import rc_util
+
+if len(sys.argv) < 2:
+    print("Usage: {} USERNAME [FULL_NAME] [REASON]".format(sys.argv[0]), file=sys.stderr)
+    exit(1)
+
+user_name = sys.argv[1]
+full_name = sys.argv[2] if len(sys.argv) >= 3 else ''
+reason    = sys.argv[3] if len(sys.argv) >= 4 else ''
+
+rc_util.add_account(user_name, full=full_name, reason=reason)
+print("Account requested for user: {}".format(user_name))
+
+print("Waiting for confirmation...")
+rc_util.consume(user_name)

rc_rmq.py

@@ -65,15 +65,16 @@ class RCRMQ(object):
                 exchange_type=self.EXCHANGE_TYPE,
                 durable=True)
 
-        if self.QUEUE is not None:
-            self._channel.queue_declare(queue=self.QUEUE, durable=self.DURABLE)
-            self._channel.queue_bind(exchange=self.EXCHANGE,
-                    queue=self.QUEUE,
-                    routing_key=self.ROUTING_KEY)
+    def bind_queue(self):
+        self._channel.queue_declare(queue=self.QUEUE, durable=self.DURABLE)
+        self._channel.queue_bind(exchange=self.EXCHANGE,
+                queue=self.QUEUE,
+                routing_key=self.ROUTING_KEY)
 
     def disconnect(self):
         self._channel.close()
         self._connection.close()
+        self._connection = None
 
     def delete_queue(self):
         self._channel.queue_delete(self.QUEUE)
@@ -89,9 +90,6 @@ class RCRMQ(object):
                 routing_key=self.ROUTING_KEY,
                 body=json.dumps(obj['msg']))
 
-        if not self._consuming:
-            self.disconnect()
-
     def start_consume(self, obj):
         if 'queue' in obj:
             self.QUEUE = obj['queue']
@@ -105,6 +103,8 @@ class RCRMQ(object):
         if self._connection is None:
             self.connect()
 
+        self.bind_queue()
+
         self._consumer_tag = self._channel.basic_consume(self.QUEUE,obj['cb'])
         self._consuming = True
         try:
@@ -112,7 +112,5 @@ class RCRMQ(object):
         except KeyboardInterrupt:
             self._channel.stop_consuming()
 
-        self.disconnect()
-
     def stop_consume(self):
         self._channel.basic_cancel(self._consumer_tag)

rc_util.py

@@ -13,6 +13,7 @@ def add_account(username, full='', reason=''):
       "reason": reason
     }
   })
+  rc_rmq.disconnect()
 
 def worker(ch, method, properties, body):
     msg = json.loads(body)
@@ -27,17 +28,18 @@ def worker(ch, method, properties, body):
             print("{} is not done yet.".format(key))
             done = False
     if done:
-        confirm_rmq.stop_consume()
-        confirm_rmq.delete_queue()
+        rc_rmq.stop_consume()
+        rc_rmq.delete_queue()
 
-def consume(username, callback, debug=False):
+def consume(username, callback=worker, debug=False):
     if debug:
         sleep(5)
     else:
-        confirm_rmq.start_consume({
+        rc_rmq.start_consume({
             'queue': username,
             'routing_key': 'confirm.' + username,
             'cb': callback
         })
+        rc_rmq.disconnect()
 
     return { 'success' : True }