ya2/engine/network/network.py

   1 # from socket import socket, AF_INET, SOCK_DGRAM, error, SOCK_STREAM, \
   2 #     SOL_SOCKET, SO_REUSEADDR
   3 # from traceback import print_exc
   4 # from logging import info
   5 # from select import select
   6 # from time import sleep
   7 # from queue import Empty
   8 # from threading import Thread
   9 # from struct import Struct, error as unpack_error
  10 # from _thread import interrupt_main
  11 # from ya2.gameobject import GameObject
  12 # from ya2.engine.network.binary import BinaryData
  13
  14
  15 # msg_rpc_call, msg_rpc_answ = range(2)
  16
  17
  18 # class _ConnectionError(Exception): pass
  19
  20
  21 # class NetworkThread(Thread):
  22
  23 #     def __init__(self, eng, port):
  24 #         Thread.__init__(self)
  25 #         self.port = port
  26 #         self.daemon = True
  27 #         self.eng = eng
  28 #         self.is_running = True
  29 #         self.size_struct = Struct('!I')
  30 #         self.tcp_sock = socket(AF_INET, SOCK_STREAM)
  31 #         self.tcp_sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
  32 #         self._configure_socket()
  33 #         self.connections = [self.tcp_sock]
  34
  35 #     def run(self):
  36 #         while self.is_running:
  37 #             sleep(.001)
  38 #             try:
  39 #                 readable, writable, exceptional = select(
  40 #                     self.connections, self.connections, self.connections, 1)
  41 #                 for sock in readable: self._process_read(sock)
  42 #                 for sock in writable: self._process_write(sock)
  43 #                 for sock in exceptional: print('exception', sock.getpeername())
  44 #             except (error, AttributeError) as exc: print_exc()
  45 #             # AttributeError happens when the server user exits from a race,
  46 #             # then destroy is being called but _process_read is still alive
  47 #             # and self.eng.cb_mux.add_cb is invoked, but self.eng in None
  48 #             except Exception as exc:
  49 #                 print_exc()
  50 #                 interrupt_main()
  51
  52 #     def _process_read(self, sock):
  53 #         try:
  54 #             data = self.recv_one_msg(sock)
  55 #             if data:
  56 #                 try:
  57 #                     msg = BinaryData.unpack(data)
  58 #                     if msg[0] == msg_rpc_call:
  59 #                         funcname, args, kwargs = msg[1:]
  60 #                         self._rpc_cb(funcname, args, kwargs, sock)
  61 #                     elif msg[0] == msg_rpc_answ:
  62 #                         self._rpc_cb(msg[1], sock)
  63 #                     else:
  64 #                         args = [msg, sock]
  65 #                         self.eng.cb_mux.add_cb(self.read_cb, args)
  66 #                 except unpack_error as exc:
  67 #                     print(exc)
  68 #                     print_exc()
  69 #         except (_ConnectionError, TypeError) as exc:
  70 #             print_exc()
  71 #             self.notify('on_disconnected', sock)
  72 #             self.connections.remove(sock)
  73
  74 #     def _process_write(self, sock):
  75 #         try:
  76 #             msg_size, msg_data = self._queue(sock).get_nowait()
  77 #             sock.sendall(self.size_struct.pack(msg_size))
  78 #             sock.sendall(msg_data)
  79 #         except Empty: pass
  80
  81 #     def recv_one_msg(self, sock):
  82 #         lengthbuf = self.recvall(sock, self.size_struct.size)
  83 #         try: length = self.size_struct.unpack(lengthbuf)[0]
  84 #         except unpack_error as exc:
  85 #             print(exc)
  86 #             raise _ConnectionError()
  87 #         return self.recvall(sock, length)
  88
  89 #     @staticmethod
  90 #     def recvall(sock, cnt):
  91 #         buf = b''
  92 #         while cnt:
  93 #             newbuf = sock.recv(cnt)
  94 #             if not newbuf: return None
  95 #             buf, cnt = buf + newbuf, cnt - len(newbuf)
  96 #         return buf
  97
  98 #     def destroy(self):
  99 #         self.is_running = False
 100 #         self.tcp_sock.close()
 101 #         self.eng = self.tcp_sock = self.connections = None
 102
 103
 104 # class AbsNetwork(GameObject):
 105
 106 #     rate = .1
 107 #     _public_addr = None
 108 #     _local_addr = None
 109
 110 #     def __init__(self, port):
 111 #         GameObject.__init__(self)
 112 #         self.netw_thr = self.read_cb = self.udp_sock = self.tcp_sock = \
 113 #             self.udp_sock = None
 114 #         self.port = port
 115 #         self.addr2conn = {}
 116
 117 #     def start(self, read_cb):
 118 #         self.eng.attach_obs(self.on_frame, 1)
 119 #         self.read_cb = read_cb
 120 #         self.udp_sock = socket(AF_INET, SOCK_DGRAM)
 121 #         self.udp_sock.setblocking(0)
 122 #         self._configure_udp()
 123 #         try:
 124 #             self.netw_thr = self._bld_netw_thr()
 125 #             self.netw_thr.start()
 126 #             self.netw_thr.read_cb = read_cb
 127 #             args = self.__class__.__name__, self.port
 128 #             info('%s is up, port %s' % args)
 129 #             return True
 130 #         except ValueError:  # e.g. empty server
 131 #             info("can't start the network")
 132
 133 #     def register_cb(self, callback):
 134 #         self.read_cb = callback
 135 #         self.netw_thr.read_cb = callback
 136
 137 #     def send(self, data_lst, receiver=None):
 138 #         dgram = BinaryData.pack(data_lst)
 139 #         self.netw_thr.send_msg(dgram, receiver)
 140
 141 #     def on_frame(self): self.process_udp()
 142
 143 #     @property
 144 #     def is_active(self):
 145 #         observers = self.eng.event.observers.values()
 146 #         return self.on_frame in [obs.mth for olst in observers for obs in olst]
 147
 148 #     def stop(self):
 149 #         if not self.netw_thr:
 150 #             info('%s was already stopped' % self.__class__.__name__)
 151 #             return
 152 #         self.udp_sock.close()
 153 #         self.netw_thr.destroy()
 154 #         self.udp_sock = self.tcp_sock = self.netw_thr = None
 155 #         self.eng.detach_obs(self.on_frame)
 156 #         self.addr2conn = {}
 157 #         info('%s has been stopped' % self.__class__.__name__)
 158
 159 #     def process_udp(self):
 160 #         try: dgram, conn = self.udp_sock.recvfrom(8192)
 161 #         except error: return
 162 #         self.on_udp_pck(dgram, conn)
 163 #         dgram = BinaryData.unpack(dgram)
 164 #         sender, payload = dgram[0], dgram[1:]
 165 #         self.read_cb(payload, conn)
 166
 167 #     def on_udp_pck(self, dgram, conn): pass
 168
 169 #     def destroy(self):
 170 #         self.stop()
 171 #         info('%s has been destroyed' % self.__class__.__name__)
 172 #         GameObject.destroy(self)