|
- import asyncio
- import logging
- import random
-
-
- logging.basicConfig(level = logging.INFO, format = '{asctime} {levelname} {message}', style = '{')
-
-
- class FakeIRCServerProtocol(asyncio.Protocol):
- def __init__(self, loop = None):
- self.loop = loop or asyncio.get_event_loop()
-
- def send(self, data):
- logging.info(f'Send: {data!r}')
- self.transport.write(data)
-
- def connection_made(self, transport):
- logging.info(f'Connection from {transport.get_extra_info("peername")}')
- self.transport = transport
- self.connected = True
- asyncio.create_task(self.pingloop())
-
- async def pingloop(self):
- while self.connected:
- self.send(b'PING :' + str(random.randint(0, 10000)).encode('ascii') + b'\r\n')
- await asyncio.sleep(10)
-
- def data_received(self, data):
- logging.info(f'Data received: {data!r}')
-
- def connection_lost(self, exc):
- logging.info(f'Connection to {self.transport.get_extra_info("peername")} lost')
- self.connected = False
-
-
- async def main():
- loop = asyncio.get_running_loop()
- server = await loop.create_server(lambda: FakeIRCServerProtocol(), '127.0.0.1', 8888)
- async with server:
- await server.serve_forever()
-
-
- asyncio.run(main())
|