-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathcodecs_socket.py
More file actions
70 lines (53 loc) · 1.71 KB
/
codecs_socket.py
File metadata and controls
70 lines (53 loc) · 1.71 KB
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
58
59
60
61
62
63
64
65
66
67
68
69
70
import sys
import socketserver
class Echo(socketserver.BaseRequestHandler):
def handle(self) -> None:
"""Get some bytes and echo them back to the client.
There is no need to decode them, since they are not used.
"""
data = self.request.recv(1024)
self.request.send(data)
class PassThrough:
def __init__(self, other):
self.other = other
def write(self, data):
print("Writing: ", repr(data))
return self.other.write(data)
def read(self, size=-1):
print("Reading: ", end=" ")
data = self.other.read(size)
print(repr(data))
return data
def flush(self):
return self.other.flush()
def close(self):
return self.other.close()
if __name__ == "__main__":
import codecs
import socket
import threading
address = ("localhost", 0) # let the kernel assign a port
server = socketserver.TCPServer(address, Echo)
ip, port = server.server_address # what port was assigned?
t = threading.Thread(target=server.serve_forever)
t.setDaemon(True)
t.start()
# Connect to the server
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((ip, port))
# Wrap the socket with a reader and writer
read_file = s.makefile("rb")
incoming = codecs.getreader("utf-8")(PassThrough(read_file))
write_file = s.makefile("wb")
outgoing = codecs.getwriter("utf-8")(PassThrough(write_file))
# Send the data
text = "中国"
print("Sending: ", repr(text))
outgoing.write(text)
outgoing.flush()
# Receive a response
response = incoming.read()
print("Received: ", repr(response))
# Clean up
s.close()
server.socket.close()