/
__servantproxy.py
358 lines (313 loc) · 12.1 KB
/
__servantproxy.py
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
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# filename: __servantproxy.py
# Tencent is pleased to support the open source community by making Tars available.
#
# Copyright (C) 2016THL A29 Limited, a Tencent company. All rights reserved.
#
# Licensed under the BSD 3-Clause License (the "License"); you may not use this file except
# in compliance with the License. You may obtain a copy of the License at
#
# https://opensource.org/licenses/BSD-3-Clause
#
# 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.
#
'''
@version: 0.01
@brief: rpc抽离出servantproxy
'''
import threading
import time
from __logger import tarsLogger
from __util import util
from __packet import RequestPacket
# from __packet import ResponsePacket
from __TimeoutQueue import ReqMessage
import exception
from exception import TarsException
class ServantProxy(object):
'''
@brief: 1、远程对象的本地代理
2、同名servant在一个通信器中最多只有一个实例
3、防止和用户在Tars中定义的函数名冲突,接口以tars_开头
'''
# 服务器响应的错误码
TARSSERVERSUCCESS = 0 # 服务器端处理成功
TARSSERVERDECODEERR = -1 # 服务器端解码异常
TARSSERVERENCODEERR = -2 # 服务器端编码异常
TARSSERVERNOFUNCERR = -3 # 服务器端没有该函数
TARSSERVERNOSERVANTERR = -4 # 服务器端五该Servant对象
TARSSERVERRESETGRID = -5 # 服务器端灰度状态不一致
TARSSERVERQUEUETIMEOUT = -6 # 服务器队列超过限制
TARSASYNCCALLTIMEOUT = -7 # 异步调用超时
TARSPROXYCONNECTERR = -8 # proxy链接异常
TARSSERVERUNKNOWNERR = -99 # 服务器端未知异常
TARSVERSION = 1
TUPVERSION = 2
TUPVERSION2 = 3
TARSNORMAL = 0
TARSONEWAY = 1
TARSMESSAGETYPENULL = 0
TARSMESSAGETYPEHASH = 1
TARSMESSAGETYPEGRID = 2
TARSMESSAGETYPEDYED = 4
TARSMESSAGETYPESAMPLE = 8
TARSMESSAGETYPEASYNC = 16
mapcls_context = util.mapclass(util.string, util.string)
def __init__(self):
tarsLogger.debug('ServantProxy:__init__')
self.__reactor = None
self.__object = None
self.__initialize = False
def __del__(self):
tarsLogger.debug('ServantProxy:__del__')
def _initialize(self, reactor, obj):
'''
@brief: 初始化函数,需要调用才能使用ServantProxy
@param reactor: 网络管理的reactor实例
@type reactor: FDReactor
@return: None
@rtype: None
'''
tarsLogger.debug('ServantProxy:_initialize')
assert(reactor and obj)
if self.__initialize:
return
self.__reactor = reactor
self.__object = obj
self.__initialize = True
def _terminate(self):
'''
@brief: 不再使用ServantProxy时调用,会释放相应资源
@return: None
@rtype: None
'''
tarsLogger.debug('ServantProxy:_terminate')
self.__object = None
self.__reactor = None
self.__initialize = False
def tars_name(self):
'''
@brief: 获取ServantProxy的名字
@return: ServantProxy的名字
@rtype: str
'''
return self.__object.name()
def tars_timeout(self):
'''
@brief: 获取超时时间,单位是ms
@return: 超时时间
@rtype: int
'''
# 默认的为3S = ObjectProxy.DEFAULT_TIMEOUT
return int(self.__timeout() * 1000)
def tars_ping(self):
pass
# def tars_initialize(self):
# pass
# def tars_terminate(self):
# pass
def tars_invoke(self, cPacketType, sFuncName, sBuffer, context, status):
'''
@brief: TARS协议同步方法调用
@param cPacketType: 请求包类型
@type cPacketType: int
@param sFuncName: 调用函数名
@type sFuncName: str
@param sBuffer: 序列化后的发送参数
@type sBuffer: str
@param context: 上下文件信息
@type context: ServantProxy.mapcls_context
@param status: 状态信息
@type status:
@return: 响应报文
@rtype: ResponsePacket
'''
tarsLogger.debug('ServantProxy:tars_invoke, func: %s', sFuncName)
req = RequestPacket()
req.iVersion = ServantProxy.TARSVERSION
req.cPacketType = cPacketType
req.iMessageType = ServantProxy.TARSMESSAGETYPENULL
req.iRequestId = 0
req.sServantName = self.tars_name()
req.sFuncName = sFuncName
req.sBuffer = sBuffer
req.iTimeout = self.tars_timeout()
reqmsg = ReqMessage()
reqmsg.type = ReqMessage.SYNC_CALL
reqmsg.servant = self
reqmsg.lock = threading.Condition()
reqmsg.request = req
reqmsg.begtime = time.time()
# # test
reqmsg.isHash = True
reqmsg.isConHash = True
reqmsg.hashCode = 123456
rsp = None
try:
rsp = self.__invoke(reqmsg)
except exception.TarsSyncCallTimeoutException:
if reqmsg.adapter:
reqmsg.adapter.finishInvoke(True)
raise
except TarsException:
raise
except:
raise TarsException('ServantProxy::tars_invoke excpetion')
if reqmsg.adapter:
reqmsg.adapter.finishInvoke(False)
return rsp
def tars_invoke_async(self, cPacketType, sFuncName, sBuffer,
context, status, callback):
'''
@brief: TARS协议同步方法调用
@param cPacketType: 请求包类型
@type cPacketType: int
@param sFuncName: 调用函数名
@type sFuncName: str
@param sBuffer: 序列化后的发送参数
@type sBuffer: str
@param context: 上下文件信息
@type context: ServantProxy.mapcls_context
@param status: 状态信息
@type status:
@param callback: 异步调用回调对象
@type callback: ServantProxyCallback的子类
@return: 响应报文
@rtype: ResponsePacket
'''
tarsLogger.debug('ServantProxy:tars_invoke')
req = RequestPacket()
req.iVersion = ServantProxy.TARSVERSION
req.cPacketType = cPacketType if callback else ServantProxy.TARSONEWAY
req.iMessageType = ServantProxy.TARSMESSAGETYPENULL
req.iRequestId = 0
req.sServantName = self.tars_name()
req.sFuncName = sFuncName
req.sBuffer = sBuffer
req.iTimeout = self.tars_timeout()
reqmsg = ReqMessage()
reqmsg.type = ReqMessage.ASYNC_CALL if callback else ReqMessage.ONE_WAY
reqmsg.callback = callback
reqmsg.servant = self
reqmsg.request = req
reqmsg.begtime = time.time()
rsp = None
try:
rsp = self.__invoke(reqmsg)
except TarsException:
raise
except Exception:
raise TarsException('ServantProxy::tars_invoke excpetion')
if reqmsg.adapter:
reqmsg.adapter.finishInvoke(False)
return rsp
def __timeout(self):
'''
@brief: 获取超时时间,单位是s
@return: 超时时间
@rtype: float
'''
return self.__object.timeout()
def __invoke(self, reqmsg):
'''
@brief: 远程过程调用
@param reqmsg: 请求数据
@type reqmsg: ReqMessage
@return: 调用成功或失败
@rtype: bool
'''
tarsLogger.debug('ServantProxy:invoke, func: %s',
reqmsg.request.sFuncName)
ret = self.__object.invoke(reqmsg)
if ret == -2:
errmsg = ('ServantProxy::invoke fail, no valid servant,' +
' servant name : %s, function name : %s' %
(reqmsg.request.sServantName,
reqmsg.request.sFuncName))
raise TarsException(errmsg)
if ret == -1:
errmsg = ('ServantProxy::invoke connect fail,' +
' servant name : %s, function name : %s, adapter : %s' %
(reqmsg.request.sServantName,
reqmsg.request.sFuncName,
reqmsg.adapter.getEndPointInfo()))
raise TarsException(errmsg)
elif ret != 0:
errmsg = ('ServantProxy::invoke unknown fail, ' +
'Servant name : %s, function name : %s' %
(reqmsg.request.sServantName,
reqmsg.request.sFuncName))
raise TarsException(errmsg)
if reqmsg.type == ReqMessage.SYNC_CALL:
reqmsg.lock.acquire()
reqmsg.lock.wait(self.__timeout())
reqmsg.lock.release()
if not reqmsg.response:
errmsg = ('ServantProxy::invoke timeout: %d, servant name'
': %s, adapter: %s, request id: %d' % (
self.tars_timeout(),
self.tars_name(),
reqmsg.adapter.trans().getEndPointInfo(),
reqmsg.request.iRequestId))
raise exception.TarsSyncCallTimeoutException(errmsg)
elif reqmsg.response.iRet == ServantProxy.TARSSERVERSUCCESS:
return reqmsg.response
else:
errmsg = 'servant name: %s, function name: %s' % (
self.tars_name(), reqmsg.request.sFuncName)
self.tarsRaiseException(reqmsg.response.iRet, errmsg)
def _finished(self, reqmsg):
'''
@brief: 通知远程过程调用线程响应报文到了
@param reqmsg: 请求响应报文
@type reqmsg: ReqMessage
@return: 函数执行成功或失败
@rtype: bool
'''
tarsLogger.debug('ServantProxy:finished')
if not reqmsg.lock:
return False
reqmsg.lock.acquire()
reqmsg.lock.notifyAll()
reqmsg.lock.release()
return True
def tarsRaiseException(self, errno, desc):
'''
@brief: 服务器调用失败,根据服务端给的错误码抛出异常
@param errno: 错误码
@type errno: int
@param desc: 错误描述
@type desc: str
@return: 没有返回值,函数会抛出异常
@rtype:
'''
if errno == ServantProxy.TARSSERVERSUCCESS:
return
elif errno == ServantProxy.TARSSERVERDECODEERR:
raise exception.TarsServerDecodeException(
"server decode exception: errno: %d, msg: %s" % (errno, desc))
elif errno == ServantProxy.TARSSERVERENCODEERR:
raise exception.TarsServerEncodeException(
"server encode exception: errno: %d, msg: %s" % (errno, desc))
elif errno == ServantProxy.TARSSERVERNOFUNCERR:
raise exception.TarsServerNoFuncException(
"server function mismatch exception: errno: %d, msg: %s" % (errno, desc))
elif errno == ServantProxy.TARSSERVERNOSERVANTERR:
raise exception.TarsServerNoServantException(
"server servant mismatch exception: errno: %d, msg: %s" % (errno, desc))
elif errno == ServantProxy.TARSSERVERRESETGRID:
raise exception.TarsServerResetGridException(
"server reset grid exception: errno: %d, msg: %s" % (errno, desc))
elif errno == ServantProxy.TARSSERVERQUEUETIMEOUT:
raise exception.TarsServerQueueTimeoutException(
"server queue timeout exception: errno: %d, msg: %s" % (errno, desc))
elif errno == ServantProxy.TARSPROXYCONNECTERR:
raise exception.TarsServerQueueTimeoutException(
"server connection lost: errno: %d, msg: %s" % (errno, desc))
else:
raise exception.TarsServerUnknownException(
"server unknown exception: errno: %d, msg: %s" % (errno, desc))