Merge branch 'rabbitmq-base-poducer-consumer' into 'version-1b-local-rabbitmq'

Refine Producer/Consumers Data Handling

See merge request mmoo97/flask_user_reg!30

README.md

@@ -21,6 +21,7 @@ $ pip install -r requirements.txt
 ```
 - Note, to install flask in your own `$HOME` use `pip install --user Flask`.
 ### RabbitMQ
+(Reference [Tutorial Part Four](https://www.rabbitmq.com/tutorials/tutorial-four-python.html))
 - Install RabbitMQ server on the host machine. (Tutorials [here](https://www.rabbitmq.com/download.html).)
   - Additionally, if on a Mac, it is recommended that you add the following line to your `.bash_profile`:  
  `export PATH=$PATH:/usr/local/opt/rabbitmq/sbin`.

base_consumer.py

 #!/usr/bin/env python
-import pika
+import pika # python client
 import sys
 
 connection = pika.BlockingConnection(
-    pika.ConnectionParameters(host='localhost'))
+    pika.ConnectionParameters(host='localhost'))  # connecting to a broker on the local machine
 channel = connection.channel()
 
-channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
+channel.exchange_declare(exchange='direct_logs', exchange_type='direct')  # create exchange to pass messages
 
 result = channel.queue_declare(queue='', exclusive=True)
-queue_name = result.method.queue
+queue_name = result.method.queue  # creates a random name for the newly generated queue
 
-severities = sys.argv[1:]
-if not severities:
-    sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])
+nodes = sys.argv[1:]
+if not nodes:
+    sys.stderr.write("Usage: %s [ood] [ohpc] [manager]\n" % sys.argv[0])
     sys.exit(1)
 
-for severity in severities:
+for node in nodes:
     channel.queue_bind(
-        exchange='direct_logs', queue=queue_name, routing_key=severity)
+        exchange='direct_logs', queue=queue_name, routing_key=node)  # combine exchange, queue, and define routing name
 
 print(' [*] Waiting for logs. To exit press CTRL+C')
 
 
 def callback(ch, method, properties, body):
     print(" [x] %r:%r" % (method.routing_key, body))
+    print('[%r]  User creation task is done.' % method.routing_key)
+
 
 
 channel.basic_consume(
-    queue=queue_name, on_message_callback=callback, auto_ack=True)
+    queue=queue_name, on_message_callback=callback, auto_ack=True)  # ingest messages, and assume delivered via auto_ack
 
-channel.start_consuming()
\ No newline at end of file
+channel.start_consuming()  # initiate message ingestion

message_manager.py

-#!/usr/bin/env python
-import pika
-
-connection = pika.BlockingConnection(
-    pika.ConnectionParameters(host='localhost'))
-channel = connection.channel()
-
-channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
-
-result = channel.queue_declare(queue='', exclusive=True)
-queue_name = result.method.queue
-
-channel.queue_bind(
-        exchange='direct_logs', queue=queue_name, routing_key="manager")
-
-print(' [*] Waiting for logs. To exit press CTRL+C')
-
-
-def callback(ch, method, properties, body):
-    print(" [x] %r:%r" % (method.routing_key, body))
-    # Todo: Make message manager more functional
-
-
-channel.basic_consume(
-    queue=queue_name, on_message_callback=callback, auto_ack=True)
-
-channel.start_consuming()
\ No newline at end of file

ohpc_consumer.py

-#!/usr/bin/env python
-import pika
-
-connection = pika.BlockingConnection(
-    pika.ConnectionParameters(host='localhost'))
-channel = connection.channel()
-
-channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
-
-result = channel.queue_declare(queue='', exclusive=True)
-queue_name = result.method.queue
-
-channel.queue_bind(
-        exchange='direct_logs', queue=queue_name, routing_key="ohpc")
-
-print(' [*] Waiting for logs. To exit press CTRL+C')
-
-
-def callback(ch, method, properties, body):
-    print(" [x] %r:%r" % (method.routing_key, body))
-
-
-channel.basic_consume(
-    queue=queue_name, on_message_callback=callback, auto_ack=True)
-
-channel.start_consuming()
\ No newline at end of file

ood_consumer.py

-#!/usr/bin/env python
-import pika
-
-connection = pika.BlockingConnection(
-    pika.ConnectionParameters(host='localhost'))
-channel = connection.channel()
-
-channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
-
-result = channel.queue_declare(queue='', exclusive=True)
-queue_name = result.method.queue
-
-channel.queue_bind(
-        exchange='direct_logs', queue=queue_name, routing_key="ood")
-
-print(' [*] Waiting for logs. To exit press CTRL+C')
-
-
-def callback(ch, method, properties, body):
-    print(" [x] %r:%r" % (method.routing_key, body))
-
-
-channel.basic_consume(
-    queue=queue_name, on_message_callback=callback, auto_ack=True)
-
-channel.start_consuming()
\ No newline at end of file

requirements.txt

@@ -47,6 +47,7 @@ six==1.13.0
 SQLAlchemy==1.3.11
 SQLAlchemy-Utils==0.35.0
 stevedore==1.31.0
+supervisor==4.1.0
 urllib3==1.25.7
 URLObject==2.4.3
 virtualenv==16.7.7

test_producer.py

@@ -8,13 +8,10 @@ channel = connection.channel()
 
 channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
 
-message = "Hey there"  # todo: account info goes here
-
+node = sys.argv[1] if len(sys.argv) > 1 else 'info'
+message = ' '.join(sys.argv[2:]) or 'Hello World!'
 channel.basic_publish(
-    exchange='direct_logs', routing_key="ohpc", body=message)
-print(" [x] Sent %r:%r" % ("ohpc", message))
+    exchange='direct_logs', routing_key=node, body=message)
+print(" [x] Sent %r:%r" % (node, message))
 
-channel.basic_publish(
-    exchange='direct_logs', routing_key="ood", body=message)
-print(" [x] Sent %r:%r" % ("ood", message))
-connection.close()
\ No newline at end of file
+connection.close()