How to use the aiomas.rpc.open_connection function in aiomas

To help you get started, we’ve selected a few aiomas examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github gdassori / spruned / spruned / services / electrod_service.py View on Github external
async def call(self, method, payload: Dict=None):
        try:
            async with async_timeout.timeout(5):
                rpc_con = await aiomas.rpc.open_connection(self.socketfile)
                call = getattr(rpc_con.remote, method)
                resp = payload is not None and await call(payload) or await call()
                await rpc_con.close()
                return resp
        except asyncio.TimeoutError:
            return

aiomas

Asyncio-based, layered networking library providing request-reply channels, RPC, and multi-agent systems.

MIT
Latest version published 7 years ago

Package Health Score

42 / 100
Full package analysis