o2ims/service/command/notify_handler.py

   1 # Copyright (C) 2021-2022 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 requests
  18 from urllib.parse import urlparse
  19
  20 # from o2common.config import config
  21 from o2common.service.unit_of_work import AbstractUnitOfWork
  22 from o2ims.domain import commands
  23 from o2ims.domain.subscription_obj import Subscription, Message2SMO
  24 from o2common.service.command.handler import get_https_conn_default
  25 from o2common.service.command.handler import get_http_conn
  26 from o2common.service.command.handler import get_https_conn_selfsigned
  27 from o2common.service.command.handler import post_data
  28 import ssl
  29 from o2common.helper import o2logging
  30 logger = o2logging.get_logger(__name__)
  31
  32
  33 # # Maybe another MQ server
  34 # r = redis.Redis(**config.get_redis_host_and_port())
  35
  36
  37 def notify_change_to_smo(
  38     cmd: commands.PubMessage2SMO,
  39     uow: AbstractUnitOfWork,
  40 ):
  41     logger.info('In notify_change_to_smo')
  42     data = cmd.data
  43     with uow:
  44         subs = uow.subscriptions.list()
  45         for sub in subs:
  46             sub_data = sub.serialize()
  47             logger.debug('Subscription: {}'.format(sub_data['subscriptionId']))
  48
  49             try:
  50                 resource_filter = json.loads(sub_data['filter'])
  51                 if len(resource_filter) > 0:
  52                     resource = uow.resources.get(data.id)
  53                     logger.debug(type(resource))
  54                     if resource:  # TODO deal with resource is empty
  55                         res_type_id = resource.serialize()['resourceTypeId']
  56                         resourcetype = uow.resource_types.get(res_type_id)
  57                         logger.debug(resourcetype.name)
  58                         if resourcetype.name not in resource_filter:
  59                             continue
  60             except json.decoder.JSONDecodeError as err:
  61                 logger.warning(
  62                     'subscription filter decode json failed: {}'.format(err))
  63
  64             callback_smo(sub, data)
  65
  66
  67 def callback_smo(sub: Subscription, msg: Message2SMO):
  68     sub_data = sub.serialize()
  69     callback_data = json.dumps({
  70         'consumerSubscriptionId': sub_data['consumerSubscriptionId'],
  71         'notificationEventType': msg.notificationEventType,
  72         'objectRef': msg.objectRef,
  73         'updateTime': msg.updatetime
  74     })
  75     logger.info('URL: {}, data: {}'.format(
  76         sub_data['callback'], callback_data))
  77     # r.publish(sub_data['subscriptionId'], json.dumps({
  78     #     'consumerSubscriptionId': sub_data['consumerSubscriptionId'],
  79     #     'notificationEventType': msg.notificationEventType,
  80     #     'objectRef': msg.objectRef
  81     # }))
  82     # try:
  83     #     headers = {'User-Agent': 'Mozilla/5.0'}
  84     #     resp = requests.post(sub_data['callback'], data=callback_data,
  85     #                          headers=headers)
  86     #     if resp.status_code == 202 or resp.status_code == 200:
  87     #         logger.info('Notify to SMO successed')
  88     #         return
  89     #     logger.error('Response code is: {}'.format(resp.status_code))
  90     # except requests.exceptions.HTTPError as err:
  91     #     logger.error('request smo error: {}'.format(err))
  92     o = urlparse(sub_data['callback'])
  93     if o.scheme == 'https':
  94         conn = get_https_conn_default(o.netloc)
  95     else:
  96         conn = get_http_conn(o.netloc)
  97     try:
  98         rst, status = post_data(conn, o.path, callback_data)
  99         if rst is True:
 100             logger.info(
 101                 'Notify to SMO successed with status: {}'.format(status))
 102             return
 103         logger.error('Notify Response code is: {}'.format(status))
 104     except ssl.SSLCertVerificationError as e:
 105         logger.debug(
 106             'Notify try to post data with trusted ca failed: {}'.format(e))
 107         if 'self signed' in str(e):
 108             conn = get_https_conn_selfsigned(o.netloc)
 109             try:
 110                 return post_data(conn, o.path, callback_data)
 111             except Exception as e:
 112                 logger.info(
 113                     'Notify post data with self-signed ca \
 114                     failed: {}'.format(e))
 115                 # TODO: write the status to extension db table.
 116                 return False
 117         return False
 118     except Exception as e:
 119         logger.critical('Notify except: {}'.format(e))
 120         return False