基于python的searpc实现（2）

在上一篇博客中，我们观察了基于python的searpc的关键机制，即searpc-server,searpc-client与客户端传输函数,在本篇博客中我们在观察一下其实现细节.

传输机制

我们已经知道,对于一个rpc框架来说,其客户端与服务器的传输机制是非常重要的.我们通过观察test_pysearpc.py文件来了解其传输机制与具体调用过程.

在这个文件中,分别定义了普通传输方式的和基于named-pipe的searpc.

searpc-server

我们首先来观察searpc-server.在之前的分析中,我们已经知道了,pysearpc这个库已经为用户准备好了一个服务器对象,其定义在server.py文件中

searpc_server = SearpcServer()

而对于普通的searpc传输,只需要import后即可使用

from pysearpc import (
    NamedPipeClient, NamedPipeServer, SearpcClient, SearpcError,
    SearpcTransport, searpc_func, searpc_server
)
....
SVCNAME = 'test-service'

def init_server():
    searpc_server.create_service(SVCNAME)
    searpc_server.register_function(SVCNAME, add, 'add')
    searpc_server.register_function(SVCNAME, mul, 'multi')
    searpc_server.register_function(SVCNAME, json_func, 'json_func')
    searpc_server.register_function(SVCNAME, get_str, 'get_str')
    
def json_func(a, b):
    return {'a': a, 'b': b}

def get_str():
    return u'这是一个测试'

只需要调用searpc_server.create_service()与searpc_server.register_function()这两个方法,即可完成创建服务与注册函数过程.至此,服务器端的工作基本完成,只需要等待客户端调用rpc函数即可.

基于`named-pipe`的searpc-server

对基于named-pipe的searpc-server的创建过程如下

SOCKET_PATH = '/tmp/libsearpc-test.sock'


class SearpcTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        init_server()
        cls.client = DummyRpcClient()

        cls.named_pipe_server = NamedPipeServer(SOCKET_PATH)
        cls.named_pipe_server.start()
        cls.named_pipe_client = NamedPipeClientForTest(SOCKET_PATH, SVCNAME)

只需创建对象并指定SOCKET_PATH即可,这是因为在pysearpc中,NamedPipeServer是对于刚刚提到的searpc_server的一个封装,其内部依然是通过searpc_server.call_function()完成的rpc函数的调用过程.可以在named_pipe.py中看到

class PipeHandlerThread(Thread):
    def __init__(self, pipe):
        Thread.__init__(self)
        self.setDaemon(True)
        self.pipe = pipe

    def run(self):
        while True:
            req_header = recvall(self.pipe, 4)
            # logger.info('Got req header %s', req_header)
            req_size, = struct.unpack('I', req_header)
            # logger.info('req size is %s', req_size)
            req = recvall(self.pipe, req_size)
            # logger.info('req is %s', req)

            data = json.loads(req.decode(encoding='utf-8'))
            resp = searpc_server.call_function(data['service'], data['request'])
            # logger.info('resp is %s', resp)

            resp_header = struct.pack('I', len(resp))
            sendall(self.pipe, resp_header)
            sendall(self.pipe, resp.encode(encoding='utf-8'))

而NamedPipeTransport类的主要作用是通过named-pipe传输数据,包括开辟线程等过程,具体见于之前的分析.

searpc-client

普通的searpc-client的传输函数与类定义如下

class DummyTransport(SearpcTransport):
    def connect(self):
        pass

    def send(self, service, fcall_str):
        return searpc_server.call_function(service, fcall_str)
        
class RpcMixin(object):
    @searpc_func("int", ["int", "int"])
    def add(self, x, y):
        pass

    @searpc_func("string", ["string", "int"])
    def multi(self, x, y):
        pass

    @searpc_func("json", ["string", "int"])
    def json_func(self, x, y):
        pass

    @searpc_func("string", [])
    def get_str(self):
        pass

class DummyRpcClient(SearpcClient, RpcMixin):
    def __init__(self):
        self.transport = DummyTransport()

    def call_remote_func_sync(self, fcall_str):
        return self.transport.send(SVCNAME, fcall_str)

由于本测试文件简化了传输过程,将searpc-client与searpc-server定义在同一片空间,因此实际上可以直接通过访问searpc_server调用到rpc函数,也并不需要连接过程.

然后可以看到DummyRpcClient的定义中,只需要指定transport属性,即传输函数,然后实现调用方法(即传输类的send()函数)即可.

最后我们看RpcMixin类,其作用是在客户端注册rpc函数,并指定返回值和参数列表.通过注册,程序员可以直接通过调用成员函数的方式调用rpc函数,可以在SearpcTest中看到其效果.

基于`named-pipe`的searpc-client

与基于named-pipe的searpc-server类似

class NamedPipeClientForTest(NamedPipeClient, RpcMixin):
    pass
...
        cls.named_pipe_client = NamedPipeClientForTest(SOCKET_PATH, SVCNAME)

但是之所以基于named-pipe的searpc-client如此简单,是因为NamedPipeClient类中已经定义了传输函数,因此程序员只需要指定通信的SOCKET_PATH,调用的服务名SVCNAME,以及注册rpc函数即可

使用过程

class SearpcTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        init_server()
        cls.client = DummyRpcClient()

        cls.named_pipe_server = NamedPipeServer(SOCKET_PATH)
        cls.named_pipe_server.start()
        cls.named_pipe_client = NamedPipeClientForTest(SOCKET_PATH, SVCNAME)

    @classmethod
    def tearDownClass(cls):
        cls.named_pipe_server.stop()

    def test_normal_transport(self):
        self.run_common(self.client)

    # @unittest.skip('not implemented yet')
    def test_pipe_transport(self):
        self.run_common(self.named_pipe_client)

    def run_common(self, client):
        v = client.add(1, 2)
        self.assertEqual(v, 3)

        v = client.multi(1, 2)
        self.assertEqual(v, 2)

        v = client.multi('abc', 2)
        self.assertEqual(v, 'abcabc')

        v = client.json_func(1, 2)
        self.assertEqual(v, json_func(1, 2))

        v = client.get_str()
        self.assertEqual(v, u'这是一个测试')

这个类定义了具体的对刚刚实现的searpc框架的测试方法.在run_common中我们可以看到,程序员可以直接通过