Broadcasting is implemented.

Author: korobool

korobool/chat/Threaded.py

@@ -13,7 +13,7 @@ def __init__(self, observer, conn, addr):
         self.name = 'no_name'
         self.observer = observer
         self.commands_queue = deque()       # each client has independent commands queue
-        self.__local_sync = threading.Lock()       # synchronization between main and client thread
+        self.__local_sync = threading.RLock()       # synchronization between main and client thread
         self.conn = conn
         self.addr = addr
         self.thread = threading.Thread(target=ServingThreadWrapper.__serve, args=(self,))
@@ -24,20 +24,25 @@ def __init__(self, observer, conn, addr):
     def send(self, data):
         with self.__local_sync:
             self.commands_queue.appendleft(data)
+            #self.__process_server_messages_queue()
 
     # Thread safe first idle message
     def post(self, data):
         with self.__local_sync:
             self.commands_queue.append(data)
+            #self.__process_server_messages_queue()
 
     # Thread safe pop command operation
     def pop_command(self):
         with self.__local_sync:
             if len(self.commands_queue) > 0:
                 return self.commands_queue.pop()
+            else:
+                return None
 
     # Thread-safe sever notifier
     def notify(self, message):
+        print('try to notify observer')
         with ServingThreadWrapper.__global_sync:
             self.observer.notify(self, message)
 
@@ -53,30 +58,35 @@ def close(self):
     @staticmethod
     def __serve(stw):
         print('Connection attempt', stw.addr)
+        stw.conn.settimeout(1)
         while not stw.closing:
             ServingThreadWrapper.__process_server_messages_queue(stw) # It might be good idea to move this to new thread
             command = ServingThreadWrapper.__receive_and_parse_client_command(stw)
             ServingThreadWrapper.__process_command(stw, command)
 
     @staticmethod
     def __process_server_messages_queue(stw):
-        package = stw.pop_command()
-        if package is None:
-            return
 
-        data = json.dumps(package)
-
-        b = bytes(data, 'utf-8')
-        stw.conn.sendall(struct.pack('i', len(data)))
+        package = stw.pop_command()
 
-        stw.conn.sendall(b)
-        print('sent data...' , b.decode('utf-8'))
+        while not package is None:
+            data = json.dumps(package)
+            b = bytes(data, 'utf-8')
+            stw.conn.sendall(struct.pack('i', len(data)))
+            stw.conn.sendall(b)
+            print('sent data...' , b.decode('utf-8'))
+            package = stw.pop_command()
 
     @staticmethod
     def __receive_and_parse_client_command(stw):
 
-        header = stw.conn.recv(4)
-        if not header: return
+        header = None
+        try:
+            header = stw.conn.recv(4)
+            if not header: return
+        except:
+            # I know that it is bad idea to swallow exceptions, this is temporary trick
+            return
 
         size = int(struct.unpack('i', header)[0])
         print('receiving bytes:', size)
@@ -99,6 +109,14 @@ def __receive_and_parse_client_command(stw):
 
     @staticmethod
     def __process_command(stw, command):
+        if command is None: return
         print('new command from client received', command['cmd'])
         if command['cmd'] == 'CMD_PING':
             stw.send({'cmd': 'CMD_PONG', 'msg': 'pong from server'})
+
+        if command['cmd'] == 'CMD_BROADCAST':
+            stw.notify(command)
+
+        if command['cmd'] == 'CMD_MESSAGE':
+            stw.notify(command)
+

korobool/chat/chat_client.py

@@ -1,3 +1,5 @@
+import threading
+
 __author__ = 'Oleksandr Korobov'
 
 import socket
@@ -9,39 +11,50 @@
 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 s.connect((HOST, PORT))
 
-def receive_and_parse_command():
 
-    header = s.recv(4)
-    if not header: return
+def receive_and_parse_command():
+    while True:
+        header = s.recv(4)
+        if not header: return
 
-    size = int(struct.unpack('i', header)[0])
-    print('receiving bytes:', size)
+        size = int(struct.unpack('i', header)[0])
+        print('receiving bytes:', size)
 
-    body = s.recv(size).decode('utf-8')
+        body = s.recv(size).decode('utf-8')
+        command = json.loads(body)
 
-    #print('!DATA!\n', body)
+        print('data from server', command)
 
-    command = json.loads(body)
+read_thread = threading.Thread(target=receive_and_parse_command)
+read_thread.start()
 
-    return command
+print("Enter commands: ")
 
 while True:
-    command_text = input("Enter command: ")
+    command_text = input()
 
     package = None
 
     if command_text.upper() == 'PING':
         package = {'cmd': 'CMD_PING'}
 
+    if command_text.upper() == 'BROADCAST':
+        message = input("Enter broadcast message: ")
+        package = {'cmd': 'CMD_BROADCAST', 'msg': message}
+
+    if command_text.upper() == 'MESSAGE':
+        print('Not supported by server yet')
+        user_id = input("Enter user id: ")
+        message = input("Enter message: ")
+        package = {'cmd': 'CMD_MESSAGE', 'msg': message}
+
     data = json.dumps(package)
 
     b = bytes(data, 'utf-8')
     s.sendall(struct.pack('i', len(data)))
 
     s.sendall(b)
 
-    #print('sent data...' , b.decode('utf-8'))
-    #s.sendall(b)
-    print(receive_and_parse_command())
 
+#read_thread.join()
 #s.close()

korobool/chat/sever.py

@@ -26,6 +26,7 @@ def __close_all_clients(self):
     def serve(self):
         self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
         self.socket.bind((self.host, self.port))
+
         while not self.closing:
             self.socket.listen(5)
             conn, addr = self.socket.accept()
@@ -34,25 +35,23 @@ def serve(self):
 
     def stop_serve(self):
         self.closing = True
-        print('Not implemented yet...')
+        print('Proper server closing  is not implemented yet...')
 
     def notify(self, sender, message):
         print('Notification received:', message)
-#        if message[0] == 'CMD_STOP':
-#            self.__close_all_clients()
-#            self.socket.close()
-#
-#        if message[0] == 'CMD_GONE':
-#            print('Not implemented yet')
-#
-#        if message[0] == 'CMD_MSGE':
-#            print('Not implemented yet')
+        if message['cmd'] == 'CMD_BROADCAST':
+            for client in self.__clients_pool:
+                if not client is None and not client.closing:
+                    #print('Posting to '. client.name)
+                    client.post(message)
+        if message['cmd'] == 'CMD_MESSAGE':
+            client.post({'cmd': 'CMD_SEVER_WARNING', 'msg': 'message sending is not implemented yet'})
 
 
 chat_sever = ChatServer(PORT = 50007)
 
 def signal_handler(signal, frame):
-    print('Wait... Server is closing now....')
+    print('CTRL+C caught. Wait... Server is closing now....')
     chat_sever.stop_serve()
     sys.exit(0)