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
171
172
173
174
|
# Copyright (c) 2019, CMCC Technologies Co., Ltd.
# 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.
import base64
import datetime
import hmac
import json
import logging
import time
from hashlib import sha1
import requests
from apscheduler.scheduler import Scheduler
from ..pub.exceptions import DmaapClientException
logger = logging.getLogger(__name__)
class BatchPublisherClient:
def __init__(self, host, topic, partition="", contenttype="text/plain", max_batch_size=100, max_batch_age_ms=1000):
self.host = host
self.topic = topic
self.partition = partition
self.contenttype = contenttype
self.max_batch_size = max_batch_size
self.max_batch_age_ms = max_batch_age_ms
self.pending = []
self.closed = False
self.dont_send_until_ms = 0
self.scheduler = Scheduler(standalone=False)
self.api_key = '',
self.api_secret = ''
@self.scheduler.interval_schedule(second=1)
def crawl_job():
self.send_message(False)
self.scheduler.start()
def set_api_credentials(self, api_key, api_secret):
self.api_key = api_key
self.api_secret = api_secret
def send(self, partition, msg):
try:
if self.closed:
raise DmaapClientException("The publisher was closed.")
message = Message(partition, msg)
self.pending.append(message)
return len(self.pending)
except Exception as e:
raise DmaapClientException("append message failed: " + e.message)
def send_message(self, force):
if force or self.should_send_now():
if not self.send_batch():
logger.error("Send failed, %s message to send.", len(self.pending))
def should_send_now(self):
should_send = False
if len(self.pending) > 0:
now_ms = int(time.time() * 1000)
should_send = len(self.pending) >= self.max_batch_size
if not should_send:
send_at_ms = self.pending[0].timestamp_ms
should_send = send_at_ms <= now_ms
should_send = should_send and now_ms >= self.dont_send_until_ms
return should_send
def send_batch(self):
if len(self.pending) < 1:
return True
now_ms = int(time.time() * 1000)
url = self.create_url()
logger.info("sending %s msgs to %s . Oldest: %s ms", len(self.pending), url,
str(now_ms - self.pending[0].timestamp_ms))
try:
str_msg = ''
if self.contenttype == "application/json":
str_msg = self.parse_json()
elif self.contenttype == "text/plain":
for m in self.pending:
str_msg += m.msg
str_msg += '\n'
elif self.contenttype == "application/cambria":
for m in self.pending:
str_msg += str(len(m.partition))
str_msg += '.'
str_msg += str(len(m.msg))
str_msg += '.'
str_msg += m.partition
str_msg += m.msg
str_msg += '\n'
else:
for m in self.pending:
str_msg += m.msg
msg = bytearray(str_msg)
start_ms = int(time.time() * 1000)
if self.api_key:
headers = self.create_headers()
else:
headers = {'content-type': self.contenttype}
ret = requests.post(url=url, data=msg, headers=headers)
if ret.status_code < 200 or ret.status_code > 299:
return False
logger.info("MR reply ok (%s ms): %s", start_ms - int(time.time() * 1000), ret.json())
self.pending = []
return True
except Exception as e:
logger.error(e.message)
return False
def create_headers(self):
data = datetime.datetime.now().strftime('%Y-%m-%dT%H:%M:%S') + '-04:00'
hmac_code = hmac.new(self.api_secret.encode(), data.encode(), sha1).digest()
signature = base64.b64encode(hmac_code).decode()
auth = self.api_key + ':' + signature
headers = {
'X-CambriaDate': data,
'X-CambriaAuth': auth,
'content-type': self.contenttype
}
return headers
def create_url(self):
url = "http://%s/events/%s" % (self.host, self.topic)
if self.partition:
url = url + "?partitionKey=" + self.partition
return url
def parse_json(self):
data = []
for message in self.pending:
msg = json.loads(message.msg)
for m in msg:
data.append(m)
return json.dumps(data)
def close(self, timeout):
try:
self.closed = True
self.scheduler.shutdown()
now_ms = int(time.time() * 1000)
wait_in_ms = now_ms + timeout * 1000
while int(time.time() * 1000) < wait_in_ms and len(self.pending) > 0:
self.send_message(True)
time.sleep(0.25)
return self.pending
except Exception as e:
raise DmaapClientException("send message failed: " + e.message)
class Message:
def __init__(self, partition, msg):
if not partition:
self.partition = ""
else:
self.partition = partition
self.msg = msg
self.timestamp_ms = int(time.time() * 1000)
|
