Initial commit

kuber.py

+import pika
+import sys
+import random
+import time
+
+
+def get_routing_key():
+    raw_data = dict(
+        first='Kuber',
+        second=['info', 'warning', 'error'],
+        third=['handle_now', 'later', 'unknown']
+    )
+    return raw_data['first'] + '.' + raw_data['second'][random.randint(0, 2)] + '.' + raw_data['third'][random.randint(0, 2)]
+
+
+connection = pika.BlockingConnection(pika.ConnectionParameters(host='nfdSS'))
+channel = connection.channel()
+
+channel.exchange_declare(exchange='topic',
+                         exchange_type='topic')
+
+num = 10 if len(sys.argv) < 2 else sys.argv[1]
+for i in range(int(num)):
+    routing_key = get_routing_key()
+    print(routing_key)
+    channel.basic_publish(exchange='topic',
+                          routing_key=routing_key,
+                          body='{}'.format(routing_key  + '@' + time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time()))))
+connection.close()

kuber_receiver.py

+import sys
+import pika
+import time
+
+
+def callback(ch, method, pro, body):
+    print("[Kuber S] Received %r" % body)
+    channel.basic_ack(delivery_tag=method.delivery_tag)
+
+
+connection = pika.BlockingConnection(pika.ConnectionParameters(host='nfdSS'))
+channel = connection.channel()
+channel.exchange_declare(exchange='topic',
+                         exchange_type='topic')
+result = channel.queue_declare('TaskQ')
+queue_name = result.method.queue
+channel.basic_qos(prefetch_count=1)
+channel.queue_bind(exchange='topic',
+                   queue=queue_name,
+                   routing_key='Task.#')
+channel.basic_consume(callback,
+                      queue=queue_name)
+channel.start_consuming()

micro.py

+import pika
+import sys
+import random
+import time
+
+
+def get_routing_key():
+    raw_data = dict(
+        first='Micro',
+        second=['import', 'convert', 'crop'],
+        third=['info', 'warning', 'error']
+    )
+    return raw_data['first'] + '.' + raw_data['second'][random.randint(0, 2)] + '.' + raw_data['third'][random.randint(0, 2)]
+
+
+connection = pika.BlockingConnection(pika.ConnectionParameters(host='nfdSS'))
+channel = connection.channel()
+
+channel.exchange_declare(exchange='topic',
+                         exchange_type='topic')
+
+num = 10 if len(sys.argv) < 2 else sys.argv[1]
+for i in range(num):
+    routing_key = get_routing_key()
+    print(routing_key)
+    channel.basic_publish(exchange='topic',
+                          routing_key=routing_key,
+                          body='{}'.format(routing_key  + '@' + time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time()))))
+connection.close()

per_receiver.py

+import sys
+import pika
+import time
+
+
+def callback(ch, method, pro, body):
+    print("[persistence] Received %r" % body)
+    # time.sleep(2)
+    channel.basic_ack(delivery_tag=method.delivery_tag)
+
+
+connection = pika.BlockingConnection(pika.ConnectionParameters(host='nfdSS'))
+channel = connection.channel()
+channel.exchange_declare(exchange='topic',
+                         exchange_type='topic')
+result = channel.queue_declare('PersistenceQ')
+queue_name = result.method.queue
+channel.basic_qos(prefetch_count=1)
+channel.queue_bind(exchange='topic',
+                   queue=queue_name,
+                   routing_key='Micro.#')
+channel.queue_bind(exchange='topic',
+                   queue=queue_name,
+                   routing_key='Kuber.#')
+channel.basic_consume(callback,
+                      queue=queue_name)
+channel.start_consuming()

seetaAtlas.py

+import pika
+import sys
+import random
+import time
+
+
+def get_routing_key():
+    raw_data = dict(
+        first='Task',
+        second=['train', 'test', 'deploy'],
+        third=['official', 'personal', 'unknown']
+    )
+    return raw_data['first'] + '.' + raw_data['second'][random.randint(0, 2)] + '.' + raw_data['third'][random.randint(0, 2)]
+
+
+connection = pika.BlockingConnection(pika.ConnectionParameters(host='nfdSS'))
+channel = connection.channel()
+
+channel.exchange_declare(exchange='topic',
+                         exchange_type='topic')
+
+num = 10 if len(sys.argv) < 2 else sys.argv[1]
+for i in range(num):
+    routing_key = get_routing_key()
+    print(routing_key)
+    channel.basic_publish(exchange='topic',
+                          routing_key=routing_key,
+                          body='{}'.format(routing_key + '@' + time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time()))))
+connection.close()

taskmaster.py

+import sys
+import pika
+import time
+
+
+def callback(ch, method, pro, body):
+    print("[taskMaster] Received %r" % body)
+    channel.basic_ack(delivery_tag=method.delivery_tag)
+
+
+connection = pika.BlockingConnection(pika.ConnectionParameters(host='nfdSS'))
+channel = connection.channel()
+channel.exchange_declare(exchange='topic',
+                         exchange_type='topic')
+result = channel.queue_declare('TaskMasterQ')
+queue_name = result.method.queue
+channel.basic_qos(prefetch_count=1)
+channel.queue_bind(exchange='topic',
+                   queue=queue_name,
+                   routing_key='#')
+channel.basic_consume(callback,
+                      queue=queue_name)
+channel.start_consuming()