Xdesk/lib/zchat.py

import socket
import sys 
import time

import select

import zlib
import base64
import json 
import time

import _thread

def dummyCB(msg):
    
    t = str(time.time())[7:]
    t = float(t)
    t1 = 0

    try:
        cmd = json.loads(msg["cmd"])
        t1 = cmd#[0]
        t1 = float(t1)
    except:pass
    print("d:",round(t1-t,3),"dummy_CB",msg)



class Server():
    def __init__(self,cb=dummyCB,port=51000):
        print("**** SERVER *****")
        self._t = time.time()
        self._last_check = time.time()
        self.port=port
        self.cb = cb
        self.clients = [] 
        self.msg=b''
        self.select = select.select

        self._start()

    def _start(self):
        self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        #self.xs.getsockopt(socket.AF_INET, socket.SO_REUSEADDR )

        while 1:
            try:
                self.server.bind(("", self.port))
                break
            except Exception as e:
                print("except",e)
                print( "bind error")
                time.sleep(1)
        
        self.server.listen(1)
        self.client_loop()

    def time(self):
        return self._t - time.time()

    def client_loop(self):
        self.client_lock = _thread.allocate_lock()
        print(dir(self.client_lock),"-----2:") # = _thread.allocate_lock()
        _thread.start_new_thread(self._client_loop,())

    def _client_loop(self):
        print("---- start server loop ----")
        while 1:
            try:
                client, addr = self.server.accept()
                client.setblocking(0)
                self.client_lock.acquire()
                self.clients.append(client)
                print("+++ Client %s open" % addr[0])
            finally:
                self.client_lock.release()
                time.sleep(0.2)
    def rem_client(self,client):
        self.client_lock.acquire()
        try:
            self.clients.remove(client)
            #print(dir(client))
            #print((client.family.name))
            print("+++ Client %s close" % client)
        finally:
            self.client_lock.release()
    def get_clients(self):
        self.client_lock.acquire()
        clients = self.clients[:]
        self.client_lock.release()
        return clients
    def _recv(self,sock):
        xmsg=b""
        msg =b""

        try:
            xmsg = sock.recv(1)#1024)#5120)
            while xmsg:
                if xmsg == b"\x00":
                    break
                msg += xmsg
                xmsg = sock.recv(1)

            idle = 0
        except ConnectionResetError as e:
            pass
        except BlockingIOError as e:
            pass
        
        if msg:
            #print("msg**",msg)
            #print("B64",sys.getsizeof(msg),len(msg))
            msg = base64.b64decode(msg)
            ##print("msg**",msg)
            ##msg = msg.decode("utf8")
            #print("str",sys.getsizeof(msg),len(msg))

            try:
                msg=zlib.decompress(msg)
                #print("uzip",sys.getsizeof(msg),len(msg))
                #print("msg",str(msg)[:150],"...")
            except Exception as e:
                print("SERVER decompress err",e)
                #msg = b"decompression error"

        
        return msg
    def check_client(self):
        if self._last_check+1 < time.time():
            self._last_check = time.time()
            for sock in self.get_clients():
                try:
                    sock.send(b".")
                except BrokenPipeError as e: 
                    self.rem_client(sock)
                except ConnectionResetError as e:
                    self.rem_client(sock)

    def poll(self):
        run = 1
        #try:
        if 1: #while run:
            self.check_client()
            idle = 1
            for sock in self.get_clients():
                #print(dir(sock))
                msg = self._recv(sock)

                if not msg:
                    continue

                idle = 0

                msg = msg.replace(b"\x00 ",b"")
                msg = {"cmd":msg}
                self.cb(msg)

        if idle:
            time.sleep(0.02)
        
        #finally:pass
        #except KeyboardInterrupt:
        #    print(" strg+c")
        #finally:
        #    for c in clients:
        #        print(c,"close")
        #        c.close()
        #    server.close()
        #    print("server close")
CMD = Server

def cmd(cb=dummyCB,port=51000):
    print("----cmd")
    x=CMD(cb=cb,port=port)
    while 1:
        x.poll()


class _Client():
    def __init__(self,port=51000):
        print("-----CLIENT-----")
        self.port = port
        self.connect()
        
    def connect(self,client_name="unkown"):
        self.xip = "127.0.0.1" #raw_input("IP-Adresse: ")
        self.xs = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        try:
            self.xs.connect((self.xip, self.port)) #50000))
        except ConnectionRefusedError as e:
            print("e654 ConnectionRefusedError: ", "ERR: {0} ".format(e.args) ,end="")
            print("Server nicht ereichbar/unterbrochen")
            print(self.xip,self.port)
            print(self.xs)
            time.sleep(1)
            self.connect()
        print("connected !")

    def send(self,nachricht):
        try:
            #print(sys.getsizeof(msg),len(msg))
            if "-dbg" in sys.argv:print("send",nachricht)
            if sys.getsizeof(nachricht):
                nachricht=zlib.compress(nachricht)
            #nachricht = bytes(nachricht,"utf-8")
            nachricht = base64.b64encode(nachricht)
            self.xs.send(nachricht + b"\00")
        except socket.error as e:
            self.connect()
        time.sleep(0.0001)
    def close(self):
        self.xs.close()
    def __del__(self):
        self.close()


import _thread as thread

class XClient():
    def __init__(self,port=51000):
        self.port = port 
        self.buffer=[]
        self._buffer=[]
        self.lock = thread.allocate_lock()
        thread.start_new_thread(self.loop,())

    def loop(self):
        
        self.Client = _Client(self.port)
        while 1:
            if not self.lock.locked():
                self.lock.acquire_lock()
                try:
                    for b in self.buffer:
                        self._buffer.append(b)
                    self.buffer = []
                finally:
                    self.lock.release_lock()
            if self._buffer:
                t = time.time()
                for b in self._buffer:
                    #print("send",len(str(b)))
                    if b[1]+2 > t:
                        self.Client.send(b[0])
                self._buffer = []
            else:
                time.sleep(0.1)
    def connect(self,client_name="unknown"):
        pass

                
    def send(self,nachricht):
        #print(self,nachricht)
        
        self.lock.acquire_lock()
        try:
            self.buffer.append([nachricht,time.time()])
        finally:
            self.lock.release_lock()
#Client = _Client
Client = XClient

tcp_sender = Client






# test ----
def run_client_test(c):
    import random 
    import string
    client = c
    print(" === TEST DATA START ===")

    try:
        for i in range(100):
            x=random.choice(string.printable)
            msg=bytes("hi"+str(x*random.randint(10,9999)),"utf-8")
            print(x,sys.getsizeof(msg),len(msg))
            client.send(msg)
            time.sleep(0.01)
    except Exception as e:
        print("e",e)
    finally:
        client.close()

    try:
        client = Client()
        for i in range(100):
            x=random.choice(string.printable)
            msg=bytes(x,"ho "+str(x*random.randint(10,9999)),"utf-8")
            print(sys.getsizeof(msg),len(msg))
            msg=zlib.compress(msg)
            print(sys.getsizeof(msg),len(msg))
            client.send(msg)
            time.sleep(0.01)
    except Exception as e:
        print("e",e)
    finally:
        pass
        #client.close()
    print(" === TEST DATA END ===")



TEST_PORT=51111
# --- single app
def test_client():
    c = Client(port=TEST_PORT)
    if "test" in sys.argv: # test server/client
        run_client_test(c)

    time.sleep(1)
    while 1:
        try:
            i=""
            i = input("cmd:")
            c.send(bytes(i,"utf8"))
        except Exception as e:
            print("e445",e)

def test_server():
    server = Server(port=TEST_PORT)
    
    while 1:
        server.poll()
        time.sleep(0.00001)

def test_cmd():
    i = sys.argv.index("cmd")
    data = sys.argv[i+1]
    print( i ,data)
    data = data.encode("utf-8")
    c = Client(TEST_PORT)
    client = c
    time.sleep(0.05)
    client.send(data)
    time.sleep(0.05)


# =======================
def main():
    if "cmd" in sys.argv:
        test_cmd()
    elif "client" in sys.argv:
        test_client()
    else: 
        test_server()


if __name__ == "__main__":
    main()