import threading | |
import socket | |
from itertools import count | |
CHARS = [chr(i).encode('ascii') for i in range(32, 127)] | |
NEWLINE = '\r\n'.encode('ascii') | |
PORT = 19 | |
WIDTH = 72 | |
def generate(client): | |
for line in count(1): | |
try: | |
for column in range(WIDTH): | |
char = CHARS[(line + column) % len(CHARS)] | |
client.send(char) | |
client.send(NEWLINE) | |
except Exception: | |
client.close() | |
break | |
def server(): | |
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) | |
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True) | |
sock.bind(('', PORT)) | |
sock.listen() | |
while True: | |
client, _ = sock.accept() | |
thread = threading.Thread(target=generate, args=(client,)) | |
thread.start() | |
if __name__ == '__main__': | |
server() |
import socket | |
from itertools import count | |
BUFFER_SIZE = 4096 | |
CHARS = [chr(i).encode('ascii') for i in range(32, 127)] | |
MAX = 511 | |
NEWLINE = '\r\n'.encode('ascii') | |
PORT = 19 | |
WIDTH = 72 | |
def generate(): | |
lines = count(1) | |
reply = b'' | |
while len(reply) < MAX: | |
line = next(lines) | |
for column in range(WIDTH): | |
reply += CHARS[(line + column) % len(CHARS)] | |
reply += NEWLINE | |
return reply[:MAX] | |
def server(): | |
reply = generate() | |
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) | |
sock.bind(('', PORT)) | |
while True: | |
_, address = sock.recvfrom(BUFFER_SIZE) | |
sock.sendto(reply, address) | |
if __name__ == '__main__': | |
server() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment