d4c7c659b2033b42e3e6d542a9083ae03b5ca23b
[pti/o2.git] / o2app / entrypoints / redis_eventconsumer.py
1 # Copyright (C) 2021 Wind River Systems, Inc.
2 #
3 #  Licensed under the Apache License, Version 2.0 (the "License");
4 #  you may not use this file except in compliance with the License.
5 #  You may obtain a copy of the License at
6 #
7 #      http://www.apache.org/licenses/LICENSE-2.0
8 #
9 #  Unless required by applicable law or agreed to in writing, software
10 #  distributed under the License is distributed on an "AS IS" BASIS,
11 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 #  See the License for the specific language governing permissions and
13 #  limitations under the License.
14
15 # import json
16 import redis
17 import json
18 from o2app import bootstrap
19 from o2common.config import config
20 # from o2common.domain import commands
21 from o2dms.domain import commands
22 from o2ims.domain import commands as imscmd
23
24 from o2common.helper import o2logging
25 from o2ims.domain.subscription_obj import Message2SMO, NotificationEventEnum, RegistrationMessage
26 logger = o2logging.get_logger(__name__)
27
28 r = redis.Redis(**config.get_redis_host_and_port())
29
30 apibase = config.get_o2ims_api_base()
31
32
33 def main():
34     logger.info("Redis pubsub starting")
35     bus = bootstrap.bootstrap()
36     pubsub = r.pubsub(ignore_subscribe_messages=True)
37     pubsub.subscribe("NfDeploymentStateChanged")
38     pubsub.subscribe('ResourceChanged')
39     pubsub.subscribe('RegistrationChanged')
40     pubsub.subscribe('OcloudChanged')
41
42     for m in pubsub.listen():
43         try:
44             handle_dms_changed(m, bus)
45         except Exception as ex:
46             logger.warning("{}".format(str(ex)))
47             continue
48
49
50 def handle_dms_changed(m, bus):
51     logger.info("handling %s", m)
52     channel = m['channel'].decode("UTF-8")
53     if channel == "NfDeploymentStateChanged":
54         datastr = m['data']
55         data = json.loads(datastr)
56         logger.info('HandleNfDeploymentStateChanged with cmd:{}'.format(data))
57         cmd = commands.HandleNfDeploymentStateChanged(
58             NfDeploymentId=data['NfDeploymentId'],
59             FromState=data['FromState'],
60             ToState=data['ToState']
61         )
62         bus.handle(cmd)
63     elif channel == 'ResourceChanged':
64         datastr = m['data']
65         data = json.loads(datastr)
66         logger.info('ResourceChanged with cmd:{}'.format(data))
67         ref = apibase + '/resourcePools/' + data['resourcePoolId'] +\
68             '/resources/' + data['id']
69         cmd = imscmd.PubMessage2SMO(data=Message2SMO(
70             id=data['id'], ref=ref,
71             eventtype=data['notificationEventType'],
72             updatetime=data['updatetime']))
73         bus.handle(cmd)
74     elif channel == 'RegistrationChanged':
75         datastr = m['data']
76         data = json.loads(datastr)
77         logger.info('RegistrationChanged with cmd:{}'.format(data))
78         cmd = imscmd.Register2SMO(data=RegistrationMessage(id=data['id']))
79         bus.handle(cmd)
80     elif channel == 'OcloudChanged':
81         datastr = m['data']
82         data = json.loads(datastr)
83         logger.info('OcloudChanged with cmd:{}'.format(data))
84         if data['notificationEventType'] == NotificationEventEnum.CREATE:
85             cmd = imscmd.Register2SMO(data=RegistrationMessage(is_all=True))
86             bus.handle(cmd)
87     else:
88         logger.info("unhandled:{}".format(channel))
89
90
91 if __name__ == "__main__":
92     main()