add hello world and work queues examples

Author: yahuarlocro

README.md

@@ -1 +1,70 @@
 # flask-rabbitmq
+
+http://localhost:15672
+user: guest
+password: guest
+
+
+# DEFINITIONS
+
+## Producer
+A program that sends messages
+
+## Queue
+It is the post box. Messages can only be stored in a queue. It is bound by
+the host's memory and disks limits. It is just a large message buffer
+
+## Consumer
+A program that wait to receive messages
+
+
+
+# HELLO WORLD
+
+RabittMQ is a message broker system. RabbitMq is a post box, a post office,
+and a letter carrier
+
+A message must always be sent by a ***exchange***. Cannot be sent direct
+to a queue
+
+
+# WORK QUEUES
+
+A work queue is used to distribute time-consuming tasks among workers
+
+The idea of work queues is to avoid waiting for a task to complete. Instead
+tasks is scheduled to be completed later
+
+We encapsulate the task as a message and send it to the queue. A worker
+running on the background (or more) will execute the job
+
+This concept is especially useful in web applications where it's impossible
+to handle a complex task during a short HTTP request window.
+
+## Message Acknowlegment
+
+What happens if a workers dies or is stopped ? Messages dispatched and
+processed by this worker are lost
+
+If a worker dies, we want the message to be delivered to another worker.
+
+In order to be sure that messages are not lost, RabbitMQ supports ***message 
+Acknowlegment***. An ack(nowledgement) is sent back by the consumer to tell
+RabbitMQ that a particular message had been received, processed and that
+RabbitMQ is free to delete it.
+
+If a worker dies (its channel is closed, connection is closed, or TCP
+connection is lost) without sending an acknowledgment, RabbitMQ understands
+that message was not processed and it is delivered to another worker (consumer)
+
+A timeout (30 minutes by default) is enforced on consumer delivery
+acknowledgement. This helps detect buggy (stuck) consumers that never 
+acknowledge deliveries. You can increase this timeout
+
+Manual message acknowledgments are turned on by default. In previous examples
+we explicitly turned them off via the ***auto_ack=True*** flag. It's time to
+remove this flag and send a proper acknowledgment from the worker, once we're
+done with a task.
+
+Acknowlegments should be sent to the same channel, otherwise will result in a 
+channel-level protocol exception

app.py

@@ -5,7 +5,7 @@
 app = Flask(__name__)
 
 
-queue_name = 'new_task_queue'
+queue_name = 'my_task_queue'
 
 @app.route('/')
 def index():
@@ -25,4 +25,4 @@ def send_order(service):
 
 
 if __name__ == '__main__':
-    app.run(debug=True, host='0.0.0.0', port=5000)
+    app.run(debug=True, host='0.0.0.0', port=5000)

connection.py

@@ -1,41 +1,49 @@
 import pika
 
-class RabbitMQConnection(object):
-# class RabbitMQConnection:
-    # def __init__(self, host, user, password, queue_name, port=5627):
-    def __init__(self, host, queue_name):
-        self.host = host
-        # self.user = user
-        # self.password = password
-        # self.port = port
-        self.queue_name = queue_name
- 
-    def __enter__(self):
 
-        # credentials = pika.PlainCredentials(self.user, self.password)
-        parameters = pika.ConnectionParameters(
-            host=self.host,
-            # port=self.port,
-            # credentials=credentials
-        )
-        # connection = pika.BlockingConnection(pika.ConnectionParameters(host=self.host))
-        self.connection = pika.BlockingConnection(parameters)
-        self.channel = self.connection.channel()
-        self.channel.queue_declare(queue=self.queue_name)
-        return self.channel
- 
-    def __exit__(self, exc_type, exc_value, traceback):
-        self.connection.close()
+# class RabbitMQConnection(object):
+#     # class RabbitMQConnection:
+#     # def __init__(self, host, user, password, queue_name, port=5627):
+#     def __init__(self, host, queue_name):
+#         self.host = host
+#         # self.user = user
+#         # self.password = password
+#         # self.port = port
+#         self.queue_name = queue_name
+#
+#     def __enter__(self):
+#
+#         # credentials = pika.PlainCredentials(self.user, self.password)
+#         parameters = pika.ConnectionParameters(
+#             host=self.host,
+#             # port=self.port,
+#             # credentials=credentials
+#         )
+#         # connection = pika.BlockingConnection(pika.ConnectionParameters(host=self.host))
+#         self.connection = pika.BlockingConnection(parameters)
+#         self.channel = self.connection.channel()
+#         self.channel.queue_declare(queue=self.queue_name)
+#         return self.channel
+#
+#     def __exit__(self, exc_type, exc_value, traceback):
+#         self.connection.close()
 
+# SENDING
+# stablich connection with rabbitmq server
+connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
+channel = connection.channel()
 
-            # connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
-    # channel = connection.channel()
-    # channel.queue_declare(queue='hello')
-    # channel.basic_publish(exchange='',
-    #                     routing_key='hello',
-    #                     body='Hello World!')
-    # connection.close()
+# create a quere were the message will be delivered
+channel.queue_declare(queue='hello')
 
+# declare exchange to specify exactly to which queue the message should go
+channel.basic_publish(exchange='', routing_key='hello', body='Hello World!')
 
-    # def __repr__(self):
-    #     return f"userID {self.user_id}"
+# debug
+print(" [x] Sent 'Hello World!'")
+
+# flush network buffers and confirm message was delivered
+connection.close()
+#
+# def __repr__(self):
+#     return f"userID {self.user_id}"

hello_world/receive.py

@@ -0,0 +1,40 @@
+import pika
+import sys
+import os
+
+
+def main():
+    connection = pika.BlockingConnection(
+            pika.ConnectionParameters(host='localhost'))
+
+    channel = connection.channel()
+
+    # make sure that the queue already exists
+    channel.queue_declare(queue='hello')
+
+    # meesages are received from the queue with a callback function, executed
+    # by the pika library
+    def callback(ch, method, properties, body):
+        print(f" [x] Received {body}")
+
+    # tell rabbitmq that this callback function receives messages from the
+    # queue
+    channel.basic_consume(queue='hello',
+                          on_message_callback=callback, auto_ack=True)
+
+    # never-ending loop that waits for data and runs callbacks whenever
+    # necessary
+    print(' [*] Waiting for messages. To exit press CTRL+C')
+    channel.start_consuming()
+
+
+if __name__ == '__main__':
+    # run programm and catch KeyboardInterrupt during program shutdown.
+    try:
+        main()
+    except KeyboardInterrupt:
+        print('Interrupted')
+        try:
+            sys.exit(0)
+        except SystemExit:
+            os._exit(0)

hello_world/send.py

@@ -0,0 +1,19 @@
+import pika
+
+# SENDING
+# stablich connection with rabbitmq server
+connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
+channel = connection.channel()
+
+# create a quere were the message will be delivered
+channel.queue_declare(queue='hello')
+
+# declare exchange to specify exactly to which queue the message should go
+channel.basic_publish(exchange='', routing_key='hello', body='Hello World!')
+
+# debug
+print(" [x] Sent 'Hello World!'")
+
+# flush network buffers and confirm message was delivered
+connection.close()
+#

jobs.sh

@@ -0,0 +1,12 @@
+#!/bin/bash
+#
+#curl http://localhost:5000/order/ndvi.
+#curl http://localhost:5000/order/ndvi..
+#curl http://localhost:5000/order/ndvi...
+#curl http://localhost:5000/order/ndvi....
+#curl http://localhost:5000/order/ndvi.....
+#curl http://localhost:5000/order/ndvi......
+#curl http://localhost:5000/order/ndvi.......
+#curl http://localhost:5000/order/ndvi........
+#curl http://localhost:5000/order/ndvi.........
+#curl http://localhost:5000/order/ndvi..........

ndvi_worker.py

@@ -4,23 +4,22 @@
 from connection import RabbitMQConnection
 from app import queue_name
 
-def main():
 
+def main():
     # Receiving messages from the queue is more complex. It works by
     # subscribing a callback function to a queue. Whenever we receive a
     # message, this callback function is called by the Pika library. In our
     # case this function will print on the screen the contents of the message.
     def callback(ch, method, properties, body):
         print(" [x] Received %r" % body.decode())
 
-
         # simulate a job that takes long. every point is one second
         time.sleep(body.count(b'.'))
 
         print(" [x] Done")
 
         # Message aknowledgement:
-        # send a proper acknowledgment from the worker, once we're done with a 
+        # send a proper acknowledgment from the worker, once we're done with a
         # task
         ch.basic_ack(delivery_tag = method.delivery_tag)
 
@@ -33,19 +32,19 @@ def callback(ch, method, properties, body):
     with RabbitMQConnection(host='localhost', queue_name=queue_name) as rc:
 
         # fair dispatch
-        # don't dispatch a new message to a worker until it has processed and 
+        # don't dispatch a new message to a worker until it has processed and
         # acknowledged the previous one. Instead, it will dispatch it to the
         # next worker that is not still busy.
         rc.basic_qos(prefetch_count=1)
 
         #  tell RabbitMQ that this particular callback function should receive
         # messages from our order queue
-        # when auto_ack is set to true, then no message aknowledgement will be 
+        # when auto_ack is set to true, then no message aknowledgement will be
         # sent in case a worker dies
         rc.basic_consume(queue=queue_name,
                          on_message_callback=callback)
                          #auto_ack=True)
-        
+
         print(' [*] Waiting for messages. To exit press CTRL+C')
 
         # enter a never-ending loop that waits for data and runs callbacks
@@ -61,4 +60,4 @@ def callback(ch, method, properties, body):
         try:
             sys.exit(0)
         except SystemExit:
-            os._exit(0)
+            os._exit(0)

receive.py

@@ -0,0 +1,40 @@
+import pika
+import sys
+import os
+
+
+def main():
+    connection = pika.BlockingConnection(
+            pika.ConnectionParameters(host='localhost'))
+
+    channel = connection.channel()
+
+    # make sure that the queue already exists
+    channel.queue_declare(queue='hello')
+
+    # meesages are received from the queue with a callback function, executed
+    # by the pika library
+    def callback(ch, method, properties, body):
+        print(f" [x] Received {body}")
+
+    # tell rabbitmq that this callback function receives messages from the
+    # queue
+    channel.basic_consume(queue='hello',
+                          on_message_callback=callback, auto_ack=True)
+
+    # never-ending loop that waits for data and runs callbacks whenever
+    # necessary
+    print(' [*] Waiting for messages. To exit press CTRL+C')
+    channel.start_consuming()
+
+
+if __name__ == '__main__':
+    # run programm and catch KeyboardInterrupt during program shutdown.
+    try:
+        main()
+    except KeyboardInterrupt:
+        print('Interrupted')
+        try:
+            sys.exit(0)
+        except SystemExit:
+            os._exit(0)

send.py

@@ -0,0 +1,19 @@
+import pika
+
+# SENDING
+# stablich connection with rabbitmq server
+connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
+channel = connection.channel()
+
+# create a quere were the message will be delivered
+channel.queue_declare(queue='hello')
+
+# declare exchange to specify exactly to which queue the message should go
+channel.basic_publish(exchange='', routing_key='hello', body='Hello World!')
+
+# debug
+print(" [x] Sent 'Hello World!'")
+
+# flush network buffers and confirm message was delivered
+connection.close()
+#