class SearpcClient(object):
def call_remote_func_sync(self, fcall_str):
raise NotImplementedError()
class RpcClientBase(SearpcClient):
def __init__(self, ccnet_client_pool, service_name, retry_num=1,
is_remote=False, remote_peer_id='', req_pool=False):
SearpcClient.__init__(self)
self.pool = ccnet_client_pool
self.service_name = service_name
self.retry_num = retry_num
self.is_remote = is_remote
self.remote_peer_id = remote_peer_id
self.req_pool = req_pool
if self.is_remote and len(self.remote_peer_id) != 40:
raise ValueError("Invalid remote peer id")
def _start_service(self, client):
req_id = client.get_request_id()
req_str = self.service_name
if self.is_remote:
req_str = "remote " + self.remote_peer_id + " " + self.service_name
client.send_request(req_id, req_str)
rsp = client.read_response()
if rsp.code != "200":
raise SearpcError("Error received: %s %s (In _start_service)" % (rsp.code, rsp.code_msg))
return req_id
def _real_call(self, client, req_id, fcall_str):
client.send_update(req_id, SC_CLIENT_CALL, SS_CLIENT_CALL, fcall_str)
rsp = client.read_response()
if rsp.code == SC_SERVER_RET:
return rsp.content
elif rsp.code == SC_SERVER_MORE:
buf = rsp.content
while True:
client.send_update(req_id, SC_CLIENT_MORE,
SS_CLIENT_MORE, '')
rsp = client.read_response()
if rsp.code == SC_SERVER_MORE:
buf += rsp.content
elif rsp.code == SC_SERVER_RET:
buf += rsp.content
break
else:
raise SearpcError("Error received: %s %s (In Read More)" % (rsp.code, rsp.code_msg))
return buf
elif rsp.code == SC_PROC_DEAD:
raise DeadProcError()
else:
raise SearpcError("Error received: %s %s" % (rsp.code, rsp.code_msg))
def call_remote_func_sync(self, fcall_str):
"""Call remote function `fcall_str` and wait response."""
retried = 0
while True:
try:
client = self.pool.get_client()
if self.req_pool:
req_id = client.req_ids.get(self.service_name, -1)
if req_id == -1:
req_id = self._start_service(client)
client.req_ids[self.service_name] = req_id
try:
ret = self._real_call(client, req_id, fcall_str)
except DeadProcError:
client.req_ids[self.service_name] = -1
self.pool.return_client(client)
if retried < self.retry_num:
retried = retried + 1
continue
else:
raise
self.pool.return_client(client)
return ret
else:
# no req pool
req_id = self._start_service(client)
ret = self._real_call(client, req_id, fcall_str)
client.send_update(req_id, "103", "service is done", "")
self.pool.return_client(client)
return ret
except (NetworkError, SearpcError):
# the client is not returned to the pool and is freed automatically
if retried < self.retry_num:
retried = retried + 1
continue
else:
raise
self.pool = ccnet_client_pool
self.service_name = service_name
self.retry_num = retry_num
self.is_remote = is_remote
self.remote_peer_id = remote_peer_id
self.req_pool = req_pool
并没有模型，这里的self.pool，self.service_name，self.retry_num，self.is_remote，self.remote_peer_id，self.req_pool是什么，临时附加的？？？
谢邀
这些应该是自定义的参数，没有文档，不过从命名可以推测含义。比如 pool是连接池，remote_peer_id 是远程端口id等等。但具体用法还得问原作者。