python/pxprpc/backend.py

   1
   2
   3
   4 import asyncio
   5 import asyncio.exceptions
   6 import typing
   7 import traceback
   8
   9 import logging
  10 log1=logging.getLogger(__name__)
  11
  12 try:
  13     from . import server
  14     class TcpServer:
  15         def __init__(self,host:str,port:int):
  16             self.host=host
  17             self.port=port
  18             self.ctxs:typing.List[server.ServerContext]=[]
  19             self.funcMap=dict()
  20
  21         async def __newConnectionHandler(self,r:asyncio.StreamReader,w:asyncio.StreamWriter):
  22             try:
  23                 ctx=server.ServerContext()
  24                 self.ctxs.append(ctx)
  25                 try:
  26                     ctx.backend1(r,w)
  27                     ctx.funcMap.update(self.funcMap)
  28                     await ctx.handle()
  29                 except asyncio.exceptions.IncompleteReadError:
  30                     log1.debug('connection closed')
  31                 except Exception as exc:
  32                     log1.debug('server exception:%s',traceback.format_exc())
  33                 finally:
  34                     self.ctxs.remove(ctx)
  35             finally:
  36                 w.close()
  37
  38         async def start(self):
  39             self.srv1:asyncio.AbstractServer=await asyncio.start_server(self.__newConnectionHandler,self.host,self.port)
  40
  41         async def stop(self):
  42             for t1 in self.ctxs:
  43                 t1.running=False
  44                 t1.out2.close()
  45             self.srv1.close()
  46
  47 except ImportError:
  48     traceback.print_exc()
  49     pass
  50
  51 try:
  52     from . import client
  53     class TcpClient:
  54         def __init__(self,host:str,port:int):
  55             self.host=host
  56             self.port=port
  57
  58         async def start(self):
  59             r,w=await asyncio.open_connection(self.host,self.port)
  60             self.rpcconn=client.RpcConnection()
  61             self.rpcconn.backend1(r,w)
  62             self.runtask=asyncio.create_task(self.rpcconn.run())
  63             self.__running=True
  64
  65         async def stop(self):
  66             if self.__running:
  67                 self.rpcconn.running=False
  68                 self.runtask.cancel()
  69                 self.__running=False
  70                 await self.rpcconn.close()
  71
  72 except ImportError:
  73     traceback.print_exc()
  74     pass