1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
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:])
|
