prpc¶

Quick start¶

Simple server:

import aiohttp.web

import prpc

class Service(object):
    @prpc.method
    async def foo(self, ctx):
        request = await ctx.stream.receive()
        assert request == "does it work?"
        await ctx.stream.send("it works!")
        return True

async def rpc_handler(request):
    connection = prpc.Connection(Service())
    return await prpc.platform.ws_aiohttp.accept(connection, request)

app = aiohttp.web.Application()
app.router.add_get("/rpc", rpc_handler)
aiohttp.web.run_app(app, port=51337)

Note that all RPC methods must accept CallContext object as a first argument.

Simple client:

import asyncio

import prpc

async def main():
    connection = prpc.Connection(None)
    await prpc.platform.ws_aiohttp.connect(
        connection, "ws://127.0.0.1:51337/rpc"
    )
    print("Connected!")
    async with connection.call_bistream("foo") as call:
        await call.stream.send("does it work?")
        response = await call.stream.receive()
        assert response == "it works!"
        result = await call.result
        print("Call result:", result)
        assert result
    await connection.close()

asyncio.get_event_loop().run_until_complete(main())

Contents¶

API Reference