diff options
Diffstat (limited to 'src/orchestration/vmb_consumer.py')
| -rw-r--r-- | src/orchestration/vmb_consumer.py | 88 |
1 files changed, 88 insertions, 0 deletions
diff --git a/src/orchestration/vmb_consumer.py b/src/orchestration/vmb_consumer.py new file mode 100644 index 0000000..3d6ebe6 --- /dev/null +++ b/src/orchestration/vmb_consumer.py @@ -0,0 +1,88 @@ +import os +import pprint + +from django.conf import settings +from pulsar import Client, AuthenticationTLS +from .vmb_messages import * + +class VMBConsumer(object): + + os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'src.far_edge_ops_api.settings') + + TLS_CERT_FILE = os.path.join(settings.VMB["CERTS_PATH"], settings.VMB["TLS_CERT_FILE"]) + TLS_KEY_FILE = os.path.join(settings.VMB["CERTS_PATH"], settings.VMB["TLS_KEY_FILE"]) + TLS_TRUST_CERTS_FILE_PATH = os.path.join(settings.VMB["CERTS_PATH"], settings.VMB["TLS_TRUST_CERTS_FILE_PATH"]) + + # Each environment(Rocklin, Branchburg, AWS Non Prod/PLE/STG/PROD has its own service url + SERVICE_URL = settings.VMB['SERVICE_URL'] + #SERVICE_URL = "pulsar://localhost:6650/" + + auth = AuthenticationTLS(TLS_CERT_FILE,TLS_KEY_FILE) + + def get_topic_message(self, topic, topic_schema): + + client = Client(self.SERVICE_URL, tls_trust_certs_file_path=self.TLS_TRUST_CERTS_FILE_PATH, + tls_allow_insecure_connection=False, authentication=self.auth) + #client = Client(SERVICE_URL) + + topic = settings.VMB["VMB_TOPICS"][topic] + consumer = client.subscribe(topic=topic, subscription_name='sub-2', schema=JsonSchema(eval(topic_schema))) + pp = pprint.PrettyPrinter(indent=4) + try: + while True: + msg = consumer.receive() + ex = msg.value() + try: + print("Received message {}".format(ex)) + consumer.acknowledge(msg) + except: + # Message failed to be processed + consumer.negative_acknowledge(msg) + except KeyboardInterrupt: + pass + + client.close() + +import sys, getopt + +def main(argv): + try: + opts, args = getopt.getopt(argv,"ht:",["topic="]) + except getopt.GetoptError: + print('vmb_consumer.py -t <topic>') + sys.exit(2) + for opt, arg in opts: + if opt == '-h': + print('vmb_consumer.py -t [cluster_status|namespace|kubeconfig|image_status>') + sys.exit() + elif opt in ("-t", "--topic"): + topic = arg + print('topic is ' + topic) + + vmbConsumer = VMBConsumer() + + + if (topic == 'namespace'): + # 1. Namespace Creation + print('subscribe to namespace topic ...') + vmbConsumer.get_topic_message(topic='TOPIC_NAMESPACE_CREATION', topic_schema='NameSpaceMessage') + + elif(topic == 'cluster_status'): + # 2. Cluster Status + print('subscribe to cluster_status topic ...') + vmbConsumer.get_topic_message(topic='TOPIC_CLUSTER_STATUS', topic_schema='ClusterStatusMessage') + + elif(topic == 'kubeconfig'): + # 3. Kubeconfig Token + print('subscribe to kubeconfig topic ...') + vmbConsumer.get_topic_message(topic='TOPIC_KUBECONFIG_TOKEN', topic_schema='KubeconfigMessage') + + elif(topic == 'image_status'): + # 4. Images Status + print('subscribe to image_status topic ...') + vmbConsumer.get_topic_message(topic='TOPIC_IMAGE_STATUS', topic_schema='ImagesStatusMessage') + else: + print('unsupport topic: ' + topic) + +if __name__ == "__main__": + main(sys.argv[1:])
\ No newline at end of file |