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
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
|
# ============LICENSE_START===================================================
# Copyright (C) 2019-2020 Nordix Foundation.
# ============================================================================
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# SPDX-License-Identifier: Apache-2.0
# ============LICENSE_END=====================================================
import uuid
import requests
from requests.auth import HTTPBasicAuth
import mod.pmsh_logging as logger
class AppConfig:
def __init__(self, **kwargs):
self.aaf_creds = {'aaf_id': kwargs.get('aaf_identity'),
'aaf_pass': kwargs.get('aaf_password')}
self.cert_path = kwargs.get('cert_path')
self.key_path = kwargs.get('key_path')
self.streams_subscribes = kwargs.get('streams_subscribes')
self.streams_publishes = kwargs.get('streams_publishes')
def get_mr_sub(self, sub_name):
"""
Returns the MrSub object requested.
Args:
sub_name: the key of the subscriber object.
Returns:
MrSub: an Instance of an `MrSub` <MrSub> Object.
Raises:
KeyError: if the sub_name is not found.
"""
try:
return _MrSub(sub_name, self.aaf_creds, **self.streams_subscribes[sub_name])
except KeyError as e:
logger.debug(e)
raise
def get_mr_pub(self, pub_name):
"""
Returns the MrPub object requested.
Args:
pub_name: the key of the publisher object.
Returns:
MrPub: an Instance of an `MrPub` <MrPub> Object.
Raises:
KeyError: if the sub_name is not found.
"""
try:
return _MrPub(pub_name, self.aaf_creds, **self.streams_publishes[pub_name])
except KeyError as e:
logger.debug(e)
raise
@property
def cert_params(self):
"""
Returns the tls artifact paths.
Returns:
cert_path, key_path: the path to tls cert and key.
"""
return self.cert_path, self.key_path
class _DmaapMrClient:
def __init__(self, aaf_creds, **kwargs):
"""
A DMaaP Message Router utility class.
Sub classes should be invoked via the AppConfig.get_mr_{pub|sub} only.
Args:
aaf_creds: a dict of aaf secure credentials.
**kwargs: a dict of streams_{subscribes|publishes} data.
"""
self.topic_url = kwargs.get('dmaap_info').get('topic_url')
self.aaf_id = aaf_creds.get('aaf_id')
self.aaf_pass = aaf_creds.get('aaf_pass')
class _MrPub(_DmaapMrClient):
def __init__(self, pub_name, aaf_creds, **kwargs):
self.pub_name = pub_name
super().__init__(aaf_creds, **kwargs)
def publish_to_topic(self, event_json):
"""
Publish the event to the DMaaP Message Router topic.
Args:
event_json: the json data to be published.
Raises:
Exception: if post request fails.
"""
try:
session = requests.Session()
headers = {'content-type': 'application/json', 'x-transactionId': str(uuid.uuid1())}
response = session.post(self.topic_url, headers=headers,
auth=HTTPBasicAuth(self.aaf_id, self.aaf_pass), json=event_json,
verify=False)
response.raise_for_status()
except Exception as e:
logger.debug(e)
raise
def publish_subscription_event_data(self, subscription, xnf_name):
"""
Update the Subscription dict with xnf and policy name then publish to DMaaP MR topic.
Args:
subscription: the `Subscription` <Subscription> object.
xnf_name: the xnf to include in the event.
"""
try:
subscription_event = subscription.prepare_subscription_event(xnf_name)
self.publish_to_topic(subscription_event)
except Exception as e:
logger.debug(f'pmsh_utils.publish_subscription_event_data : {e}')
class _MrSub(_DmaapMrClient):
def __init__(self, sub_name, aaf_creds, **kwargs):
self.sub_name = sub_name
super().__init__(aaf_creds, **kwargs)
def get_from_topic(self, consumer_id, consumer_group='dcae_pmsh_cg', timeout=1000):
"""
Returns the json data from the MrTopic.
Args:
consumer_id: Within your subscribers group, a name that uniquely
identifies your subscribers process.
consumer_group: A name that uniquely identifies your subscribers.
timeout: The request timeout value in mSec.
Returns:
list[str]: the json response from DMaaP Message Router topic, else None.
"""
topic_data = None
try:
session = requests.Session()
headers = {'accept': 'application/json', 'content-type': 'application/json'}
response = session.get(f'{self.topic_url}/{consumer_group}/{consumer_id}'
f'?timeout={timeout}',
auth=HTTPBasicAuth(self.aaf_id, self.aaf_pass), headers=headers,
verify=False)
response.raise_for_status()
if response.ok:
topic_data = response.json()
except Exception as e:
logger.debug(e)
return topic_data
|