o2ims/service/command/notify_handler.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 requests
  18 import http.client
  19 from urllib.parse import urlparse
  20
  21 # from o2common.config import config
  22 from o2common.service.unit_of_work import AbstractUnitOfWork
  23 from o2ims.domain import commands
  24 from o2ims.domain.subscription_obj import Subscription, Message2SMO
  25
  26 from o2common.helper import o2logging
  27 logger = o2logging.get_logger(__name__)
  28
  29 # # Maybe another MQ server
  30 # r = redis.Redis(**config.get_redis_host_and_port())
  31
  32
  33 def notify_change_to_smo(
  34     cmd: commands.PubMessage2SMO,
  35     uow: AbstractUnitOfWork,
  36 ):
  37     logger.info('In notify_change_to_smo')
  38     data = cmd.data
  39     with uow:
  40         subs = uow.subscriptions.list()
  41         for sub in subs:
  42             sub_data = sub.serialize()
  43             logger.debug('Subscription: {}'.format(sub_data['subscriptionId']))
  44
  45             try:
  46                 resource_filter = json.loads(sub_data['filter'])
  47                 if len(resource_filter) > 0:
  48                     resource = uow.resources.get(data.id)
  49                     logger.debug(type(resource))
  50                     if resource:  # TODO deal with resource is empty
  51                         res_type_id = resource.serialize()['resourceTypeId']
  52                         resourcetype = uow.resource_types.get(res_type_id)
  53                         logger.debug(resourcetype.name)
  54                         if resourcetype.name not in resource_filter:
  55                             continue
  56             except json.decoder.JSONDecodeError as err:
  57                 logger.warning(
  58                     'subscription filter decode json failed: {}'.format(err))
  59
  60             callback_smo(sub, data)
  61
  62
  63 def callback_smo(sub: Subscription, msg: Message2SMO):
  64     sub_data = sub.serialize()
  65     callback_data = json.dumps({
  66         'consumerSubscriptionId': sub_data['consumerSubscriptionId'],
  67         'notificationEventType': msg.notificationEventType,
  68         'objectRef': msg.objectRef,
  69         'updateTime': msg.updatetime
  70     })
  71     logger.info('URL: {}, data: {}'.format(
  72         sub_data['callback'], callback_data))
  73     # r.publish(sub_data['subscriptionId'], json.dumps({
  74     #     'consumerSubscriptionId': sub_data['consumerSubscriptionId'],
  75     #     'notificationEventType': msg.notificationEventType,
  76     #     'objectRef': msg.objectRef
  77     # }))
  78     # try:
  79     #     headers = {'User-Agent': 'Mozilla/5.0'}
  80     #     resp = requests.post(sub_data['callback'], data=callback_data,
  81     #                          headers=headers)
  82     #     if resp.status_code == 202 or resp.status_code == 200:
  83     #         logger.info('Notify to SMO successed')
  84     #         return
  85     #     logger.error('Response code is: {}'.format(resp.status_code))
  86     # except requests.exceptions.HTTPError as err:
  87     #     logger.error('request smo error: {}'.format(err))
  88     o = urlparse(sub_data['callback'])
  89     conn = http.client.HTTPConnection(o.netloc)
  90     headers = {'Content-type': 'application/json'}
  91     conn.request('POST', o.path, callback_data, headers)
  92     resp = conn.getresponse()
  93     data = resp.read().decode('utf-8')
  94     # json_data = json.loads(data)
  95     if resp.status == 202 or resp.status == 200:
  96         logger.info('Notify to SMO successed, response code {} {}, data {}'.
  97                     format(resp.status, resp.reason, data))
  98         return
  99     logger.error('Response code is: {}'.format(resp.status))