blob: 697af9f22e00e246aacf2dc23a53ad2d6c76e054 (
about) (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
|
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()
self.buffer = b''
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}')
messages = data.split(b'\r\n')
if self.buffer:
self.message_received(self.buffer + messages[0])
messages = messages[1:]
for message in messages[:-1]:
self.message_received(message)
self.buffer = messages[-1]
def message_received(self, message):
if message.startswith(b'PING '):
self.send(b'PONG ' + message[5:] + b'\r\n')
elif message.startswith(b'USER '):
self.send(b'001 :Hello there!\r\n')
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())
|
