client and server commands

Author: wezs

tmarchuk/client.py

@@ -3,17 +3,17 @@
 import signal
 import socket
 import sys
+import threading
 
 
 def sig_handler(signum, frame):
     global client
-    if client and getattr(client, 'connected'):
-        client.connected = False
+    if client and getattr(client, 'stop'):
+        client.stop.set()
 
 
 class Client:
-    connected = False
-    CHUNK_SIZE = 1024
+    cmd_list = ['connected', 'pong', 'pongd', 'ackreply', 'ackfinish']
 
     def __init__(self, host='localhost', port=39999):
         self.host = host
@@ -22,30 +22,63 @@ def __init__(self, host='localhost', port=39999):
         try:
             self.socket.connect((self.host, self.port))
         except socket.error as e:
-            print("Can't connect to server: %s, %s" % (e.errno, e.strerror))
+            print("Can't connect to server: {}, {}".format(e.errno, e.strerror))
             sys.exit(2)
-        else:
-            self.connected = True
+        self.stop = threading.Event()
 
     def start(self):
-        while self.connected:
+        self.rcv_thread = threading.Thread(target=self.rcv_handler)
+        self.rcv_thread.start()
+        while not self.stop.wait(0):
             input_data = input('=> ')
             if len(input_data.strip()):
-                self.send(input_data)
-                data = self.recv()
-                print(data)
+                data = '\n'.join(input_data.split(' ', 1))
+                self.send(data)
+        self.terminate()
+
+    def terminate(self):
         self.socket.close()
+        self.rcv_thread.join()
+
+    def parse_data(self, data):
+        data_list = data.split('\n', 1)
+        return data_list[0], data_list[1:]
+
+    def rcv_handler(self):
+        while not self.stop.wait(0):
+            data = self.recv()
+            if not data:
+                print("Connection closed by server")
+                self.stop.set()
+                break
+            cmd, cmd_args = self.parse_data(data)
+            if cmd in self.cmd_list and callable(getattr(self, cmd, None)):
+                getattr(self, cmd)(cmd_args)
+            else:
+                print(data)
+
+    def connected(self, data):
+        print('New client connected from {}'.format(data))
+
+    def ackquit(self, data):
+        print('Someone has gone: "{}"'.format(data))
+
+    def ackfinish(self, data):
+        print("Server is going down!")
+        self.terminate()
 
     def send(self, data):
         snd_data = data.encode('utf-8')
+        snd_length = int(len(snd_data)).to_bytes(4, 'big')
+        self.socket.sendall(snd_length)
         self.socket.sendall(snd_data)
 
     def recv(self):
-        rcv_data = self.socket.recv(self.CHUNK_SIZE)
+        rcv_length = int.from_bytes(self.socket.recv(4), 'big')
+        rcv_data = self.socket.recv(rcv_length)
         data = rcv_data.decode('utf-8')
         return data
 
-
 if __name__ == '__main__':
     client = Client()
     signal.signal(signal.SIGINT, sig_handler)

tmarchuk/server.py

@@ -17,26 +17,24 @@ def sig_handler(signum, frame):
         server.stop()
 
 
-class Connection(threading.Thread):
-    cmd_list = []
-    CHUNK_SIZE = 1024
+class Connection:
+    cmd_list = ['connect', 'ping', 'pingd', 'quit', 'finish']
 
-    def __init__(self, conn, addr):
+    def __init__(self, conn, addr, srv):
         self.conn = conn
         self.addr = addr
+        self.srv = srv
         self.stop = threading.Event()
-        threading.Thread.__init__(self)
-
-    def terminate(self):
-        self.stop.set()
+        self.thread = threading.Thread(target=self.run)
+        self.thread.start()
 
     def run(self):
         while not self.stop.wait(0):
             try:
                 data = self.recv()
                 if data and len(data.strip()):
                     cmd, cmd_args = self.parse_data(data)
-                    if cmd in self.cmd_list and callable(getattr(self, cmd)):
+                    if cmd in self.cmd_list and callable(getattr(self, cmd, None)):
                         getattr(self, cmd)(cmd_args)
                     else:
                         self.send('No such command')
@@ -49,31 +47,59 @@ def run(self):
                 else:
                     logger.info("IO error: {}, {}".format(e.errno, e.strerror))
                 break
+        self.stop.set()
         self.conn.close()
 
+    def terminate(self):
+        self.stop.set()
+
     def send(self, data):
         logger.debug("sending: {}".format(repr(data)))
         snd_data = data.encode('utf-8')
+        snd_length = int(len(snd_data)).to_bytes(4, 'big')
+        self.conn.sendall(snd_length)
         self.conn.sendall(snd_data)
 
     def recv(self):
-        rcv_data = self.conn.recv(self.CHUNK_SIZE)
+        rcv_length = int.from_bytes(self.conn.recv(4), 'big')
+        rcv_data = self.conn.recv(rcv_length)
         if not rcv_data:
             logger.info("Connection closed by foreign host.")
-            self.terminate()
+            self.stop.set()
             return False
         data = rcv_data.decode('utf-8')
-        logger.debug("received: %s".format(repr(data)))
+        logger.debug("received: {}".format(repr(data)))
         return data
 
     def parse_data(self, data):
-        data_list = data.split()
+        data_list = data.split('\n', 1)
         return data_list[0], data_list[1:]
 
+    def connect(self, data):
+        self.srv.multicast_message('connected\n{}'.format(
+                ':'.join(map(str,  self.addr))
+            ))
+
+    def ping(self, data):
+        self.send('pong')
+
+    def pingd(self, data):
+        self.send('pongd\n{}'.format(' '.join(data)))
+
+    def quit(self, data):
+        if data:
+            self.srv.multicast_message('ackquit\n{}'.format(data))
+        self.stop.set()
+
+    def finish(self, data):
+        self.stop.set()
+        self.srv.terminate()
+
 
 class Server:
     running = False
     connections = []
+    connections_lock = threading.RLock()
 
     def __init__(self, host='', port=39999):
         logger.info("Init server at {}:{}".format(host, port))
@@ -96,20 +122,35 @@ def run(self):
         while self.running:
             try:
                 conn, addr = self.socket.accept()
-            except (socket.timeout, InterruptedError):
+            except socket.timeout:
                 pass
+            except (KeyboardInterrupt, InterruptedError):
+                self.stop()
             else:
-                th_connection = Connection(conn, addr)
-                th_connection.start()
-                self.connections.append(th_connection)
+                client_connection = Connection(conn, addr, self)
+                with self.connections_lock:
+                    self.connections.append(client_connection)
+        self.stop()
+
+    def terminate(self):
+        self.running = False
 
     def stop(self):
         self.running = False
+        self.multicast_message('ackfinish')
+        # self.ward_thread.join()
+        with self.connections_lock:
+            for conn in self.connections:
+                conn.terminate()
+                conn.thread.join()
+                del conn
         self.socket.close()
-        for conn in self.connections:
-            conn.terminate()
-            conn.join()
-            del conn
+
+    def multicast_message(self, msg):
+        with self.connections_lock:
+            for conn in self.connections:
+                if not conn.stop.wait(0):
+                    conn.send(msg)
 
 
 if __name__ == '__main__':