安装

pip install lacia>=0.2

pdm add lacia>=0.2

特性

• 跨语言
  • Python3
  • JavaScript/TypeScript
  • Golang
  • C/C++
  • Java
  • Rust
• 多种网络协议支持
  • HTTP
  • WebSocket
  • 自定义
• 多种Runtime支持
  • 兼容 Json-Rpc Ast 规范
  • 兼容 Json-Rpc 2.0 规范
  • 兼容 Json-Rpc X 规范
• 支持完备的链式调用
• 支持嵌套调用
• 支持双向调用 (StoC, CtoS, CtoC)
• 双向流式传输
• 支持 BSON
• IDE 支持
• 分布式Server

使用

入门

Server 端

from lacia.core.core import JsonRpc
from lacia.network.server.aioserver import AioServer, mount_app

namespace = {
    "ping": lambda x: f"pong {x}",
}

rpc = JsonRpc(name = "server_test", namespace=namespace)

if __name__ == "__main__":

    from aiohttp import web
    app = web.Application()
    mount_app(app=app, server=server, rpc=rpc, path="/ws")
    web.run_app(app, host="localhost", port=8080)

Client 端

import asyncio
from lacia.core.core import JsonRpc
from lacia.core.proxy import ProxyObj
from lacia.network.client.aioclient import AioClient

rpc = JsonRpc(
    name="client_test",
)

async def main():

    client = AioClient(path="/ws")
    await rpc.run_client(client)

    ping = ProxyObj(rpc).ping

    assert await ping("hello") == "pong hello"

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

链式调用

Server 端

from lacia.core.core import JsonRpc
from lacia.core.proxy import ProxyObj
from lacia.network.client.aioclient import AioClient

rpc = JsonRpc(
    name="client_test",
)

class Test:

    def __init__(self, a, b) -> None:
        self.a = a
        self.b = b
    
    def output(self, name):
        return f"hello: {name}, a: {self.a}, b: {self.b}"

namespace = {
    "Test": Test
}

rpc = JsonRpc(name = "server_test", namespace=namespace)

if __name__ == "__main__":

    from aiohttp import web
    app = web.Application()
    mount_app(app=app, server=server, rpc=rpc, path="/ws")
    web.run_app(app, host="localhost", port=8080)

Client 端

import asyncio
from lacia.core.core import JsonRpc
from lacia.core.proxy import ProxyObj
from lacia.network.client.aioclient import AioClient

rpc = JsonRpc(
    name="client_test",
)

async def main():

    client = AioClient(path="/ws")
    await rpc.run_client(client)

    Test = ProxyObj(rpc).Test

    assert await Test(1, 2).output("world") == "hello: world, a: 1, b: 2"

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

反向调用

Server 端

import asyncio
from lacia.core.core import JsonRpc
from lacia.core.proxy import ProxyObj
from lacia.network.server.aioserver import AioServer, mount_app

namespace = {
    "ping": lambda x: f"pong {x}",
}

rpc = JsonRpc(name = "server_test", namespace=namespace)

async def reverse_call(name: str):
    res = await ProxyObj(rpc, name=name).ping(name)
    return res

rpc.add_namespace({
    "reverse_call": reverse_call,
})

if __name__ == "__main__":

    from aiohttp import web
    app = web.Application()
    mount_app(app=app, server=server, rpc=rpc, path="/ws")
    web.run_app(app, host="localhost", port=8080)

Client 端

from pprint import pprint

import asyncio
from lacia.core.core import JsonRpc
from lacia.core.proxy import ProxyObj
from lacia.network.client.aioclient import AioClient

namespace = {
    "ping": lambda x: f"pong {x}",
}

rpc = JsonRpc(
    name="client_test",
    namespace=namespace
)

async def main():

    client = AioClient(path="/ws")
    await rpc.run_client(client)

    proxy1 = ProxyObj(rpc).reverse_call
    assert await proxy1("hello") == "pong hello"

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

流式调用

Server 端

import asyncio
from lacia.core.core import JsonRpc
from lacia.core.proxy import ProxyObj
from lacia.network.server.aioserver import AioServer, mount_app

async def test_async_iter(n: int):
    for i in range(n):
        await asyncio.sleep(1)
        yield i

namespace = {
    "test_async_iter": test_async_iter,
}

rpc = JsonRpc(name = "server_test", namespace=namespace)

if __name__ == "__main__":

    from aiohttp import web
    app = web.Application()
    mount_app(app=app, server=server, rpc=rpc, path="/ws")
    web.run_app(app, host="localhost", port=8080)

Client 端

from pprint import pprint

import asyncio
from lacia.core.core import JsonRpc
from lacia.core.proxy import ProxyObj
from lacia.network.client.aioclient import AioClient

namespace = {
    "ping": lambda x: f"pong {x}",
}

rpc = JsonRpc(
    name="client_test",
    namespace=namespace
)

async def main():

    client = AioClient(path="/ws")
    await rpc.run_client(client)

    proxy3 = ProxyObj(rpc).test_async_iter

    async for i in proxy3(3):
        print(i)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

反向流式调用

Server 端

import asyncio
from lacia.core.core import JsonRpc, Context
from lacia.core.proxy import ProxyObj
from lacia.network.server.aioserver import AioServer, mount_app

async def test_async_iter(n: int):
    name = Context.name.get()
    obj = ProxyObj(Context.rpc.get(), name)
    async for i in obj.test_async_iter(n):
        yield i

namespace = {
    "test_async_iter": test_async_iter,
}

rpc = JsonRpc(name = "server_test", namespace=namespace)

if __name__ == "__main__":

    from aiohttp import web
    app = web.Application()
    mount_app(app=app, server=server, rpc=rpc, path="/ws")
    web.run_app(app, host="localhost", port=8080)

Client 端

import asyncio
from lacia.core.core import JsonRpc
from lacia.core.proxy import ProxyObj
from lacia.network.client.aioclient import AioClient

async def test_async_iter(n: int):
    for i in range(n):
        await asyncio.sleep(1)
        yield i

namespace = {
    "test_async_iter": test_async_iter,
}

rpc = JsonRpc(
    name="client_test",
    namespace=namespace
)

async def main():

    client = AioClient(path="/ws")
    await rpc.run_client(client)

    proxy = ProxyObj(rpc).test_async_iter

    async for i in proxy(3):
        print(i)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

嵌套调用

Server 端

import asyncio
from lacia.core.core import JsonRpc
from lacia.core.proxy import ProxyObj
from lacia.network.server.aioserver import AioServer, mount_app

class Test:

    def __init__(self, a, b) -> None:
        self.a = a
        self.b = b
    
    def output(self, name):
        return f"hello: {name}"

def class_sum(obj: Test):
    return obj.a + obj.b

namespace = {
    "class_sum": class_sum,
    "Test": Test,
}

rpc = JsonRpc(name = "server_test", namespace=namespace)

if __name__ == "__main__":

    from aiohttp import web
    app = web.Application()
    mount_app(app=app, server=server, rpc=rpc, path="/ws")
    web.run_app(app, host="localhost", port=8080)

Client 端

from pprint import pprint

import asyncio
from lacia.core.core import JsonRpc
from lacia.core.proxy import ProxyObj
from lacia.network.client.aioclient import AioClient

namespace = {
    "ping": lambda x: f"pong {x}",
}

rpc = JsonRpc(
    name="client_test",
    namespace=namespace
)

async def main():

    client = AioClient(path="/ws")
    await rpc.run_client(client)

    proxy1 = ProxyObj(rpc).Test(1, b=2)
    proxy2 = ProxyObj(rpc).class_sum(proxy1)

    assert await proxy2 == 3

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

Client to Client

Server 端

import asyncio
from lacia.core.core import JsonRpc, Context
from lacia.core.proxy import ProxyObj
from lacia.network.server.aioserver import AioServer

namespace = {
    "ping": lambda x: f"pong {x}",
}

rpc = JsonRpc(name = "server_test")

if __name__ == "__main__":

    from aiohttp import web
    app = web.Application()
    mount_app(app=app, server=server, rpc=rpc, path="/ws")
    web.run_app(app, host="localhost", port=8080)

Client 端 A

import asyncio
from lacia.core.core import JsonRpc
from lacia.core.proxy import ProxyObj
from lacia.network.client.aioclient import AioClient

async def test_async_iter(n: int):
    for i in range(n):
        await asyncio.sleep(1)
        yield i

namespace = {
    "ping": lambda x: f"pong {x}",
    "test_async_iter": test_async_iter,
}

rpc = JsonRpc(
    name="client_test_A",
    namespace=namespace
)

async def main():

    client = AioClient(path="/ws")
    await rpc.run_client(client)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.run_forever()

Client 端 B

import asyncio
from lacia.core.core import JsonRpc
from lacia.core.proxy import ProxyObj
from lacia.network.client.aioclient import AioClient

namespace = {
    "ping": lambda x: f"pong {x}",
}

rpc = JsonRpc(
    name="client_test_B",
    namespace=namespace
)

async def main():

    client = AioClient(path="/ws")
    await rpc.run_client(client)

    c_obj = ProxyObj(rpc, "client_test_A")
    s_obj = ProxyObj(rpc)

    async for i in c_obj.test_async_iter(10):
        print(i)
    
    print(await s_obj.ping(c_obj.ping("hello")))
    print(await c_obj.ping(s_obj.ping("hello")))

loop = asyncio.get_event_loop()
loop.run_until_complete(main())