rudp server basically fin

Author: no5ix

pycharm2020.1.3/.idea/pycharm2020.1.3.iml

@@ -54,7 +54,7 @@ def __init__(
             transport: transports.BaseTransport = None
     ):
 
-        self._conn_type = None
+        self._proto_type = None
 
         self._try_connect_times = 0
 
@@ -96,6 +96,9 @@ def __init__(
         else:
             self._conn_state = CONN_STATE_CONNECTED
 
+    def get_proto_type(self):
+        return self._proto_type
+
     # @wait_or_not()
     async def try_connect(self) -> bool:
         raise NotImplementedError

pycharm2020.1.3/script/ConnBase.py

@@ -13,16 +13,16 @@
 from core.util.TimerHub import TimerHub
 from core.util.UtilApi import Singleton
 
-from core.common.sanic_jwt_extended import JWT, refresh_jwt_required, jwt_required, jwt_optional
-from core.common.sanic_jwt_extended.tokens import Token
+from sanic_jwt_extended import JWT
+from sanic_jwt_extended.tokens import Token
 
 if typing.TYPE_CHECKING:
-    from RpcHandler import RpcHandler, RPC_TYPE_REQUEST
+    from RpcHandler import RpcHandler
     import TcpConn
 
 
-CONN_TYPE_TCP = 0
-CONN_TYPE_RUDP = 1
+PROTO_TYPE_TCP = 0
+PROTO_TYPE_RUDP = 1
 
 ROLE_TYPE_ACTIVE = 0
 ROLE_TYPE_PASSIVE = 1
@@ -43,13 +43,13 @@ def connection_made(self, transport: transports.BaseTransport) -> None:
         addr = transport.get_extra_info('peername')  # type: typing.Tuple[str, int]
         if self._role_type == ROLE_TYPE_PASSIVE:
             self._conn = ConnMgr.instance().add_incoming_conn(
-                CONN_TYPE_TCP, transport, addr
+                PROTO_TYPE_TCP, transport, addr
                 # self._rpc_handler
             )
         else:
-            self._conn = ConnMgr.instance().get_conn(addr, CONN_TYPE_TCP)
+            self._conn = ConnMgr.instance().get_conn(addr, PROTO_TYPE_TCP)
             assert self._conn
-        LogManager.get_logger().info(f"{addr!r} is connected !!!!")
+        LogManager.get_logger().info(f"TCP {addr!r} is connected !!!!")
 
     def data_received(self, data: bytes) -> None:
         self._conn.handle_read(data)
@@ -70,6 +70,9 @@ class RudpProtocol(asyncio.DatagramProtocol):
     # def __init__(self, create_kcp_conn_cb):
     def __init__(self):
         # self._create_kcp_conn_cb = create_kcp_conn_cb
+        with JWT.initialize() as manager:
+            manager.config.secret_key = "new_byte"
+
         self.transport = None
 
     def connection_made(self, transport):
@@ -82,6 +85,7 @@ def datagram_received(self, data: bytes, addr):
         global RUDP_HANDSHAKE_ACK_PREFIX
         # assert callable(self._create_kcp_conn_cb)
         # self._create_kcp_conn_cb(addr)
+        # print(f'datagram_received: {data=}, {addr=}')
         if data == RUDP_HANDSHAKE_SYN:
             RUDP_CONV += 1
             access_token_jwt: bytes = JWT.create_access_token(
@@ -96,14 +100,14 @@ def datagram_received(self, data: bytes, addr):
             token_obj = Token(raw_jwt)
             if token_obj.type != "access":
                 raise Exception("Only access tokens are allowed")
-            conv = token_obj.identity
+            conv = int(token_obj.identity)
 
             ConnMgr.instance().add_incoming_conn(
-                CONN_TYPE_RUDP, self.transport, addr, rudp_conv=conv
+                PROTO_TYPE_RUDP, self.transport, addr, rudp_conv=conv
                 # self._rpc_handler
             )
             # addr = transport.get_extra_info('peername')  # type: typing.Tuple[str, int]
-            LogManager.get_logger().info(f"{addr!r} is connected !!!!")
+            LogManager.get_logger().info(f"RUDP {addr!r} is connected !!!!")
         elif data.startswith(RUDP_HANDSHAKE_SYN_ACK_PREFIX):
             parts = data.split(RUDP_HANDSHAKE_SYN_ACK_PREFIX, 2)
             if len(parts) != 2:
@@ -114,22 +118,18 @@ def datagram_received(self, data: bytes, addr):
             if token_obj.type != "access":
                 raise Exception("Only access tokens are allowed")
             self.transport.sendto(RUDP_HANDSHAKE_ACK_PREFIX + raw_jwt.encode(), addr)
-            conv = token_obj.identity
-            try:
-                ConnMgr.instance().set_fut_result(addr, conv)
-            except asyncio.InvalidStateError:
-                # 并发情况即使`fut`已经done了也无所谓, 不处理即可
-                pass
+            conv = int(token_obj.identity)
+            ConnMgr.instance().set_fut_result(addr, conv)
 
             # ConnMgr.instance().add_incoming_conn(
             #     ROLE_TYPE_ACTIVE, CONN_TYPE_RUDP, self.transport,
             #     rudp_conv=conv, rudp_peer_addr=addr
             #     # self._rpc_handler
             # )
             # addr = transport.get_extra_info('peername')  # type: typing.Tuple[str, int]
-            LogManager.get_logger().info(f"{addr!r} is connected !!!!")
+            LogManager.get_logger().info(f"RUDP {addr!r} is connected !!!!")
         else:
-            _cur_conn = ConnMgr.instance().get_conn(addr, CONN_TYPE_RUDP)
+            _cur_conn = ConnMgr.instance().get_conn(addr, PROTO_TYPE_RUDP)
             assert _cur_conn
             _cur_conn.handle_read(data)
 
@@ -141,7 +141,7 @@ def __init__(self):
         self._timer_hub = TimerHub()
         self._logger = LogManager.get_logger()
         self._is_proxy = False
-        self._conn_type_2_addr_2_conn = {CONN_TYPE_TCP: {}, CONN_TYPE_RUDP: {}}
+        self._proto_type_2_addr_2_conn = {PROTO_TYPE_TCP: {}, PROTO_TYPE_RUDP: {}}
         self._addr_2_try_connect_times = defaultdict(int)
         self._addr_2_rudp_conned_fut = {}
 
@@ -156,58 +156,68 @@ def final_fut_cb(fut, _addr=addr):
         _fut.add_done_callback(final_fut_cb)
         return _fut
 
+    def set_fut_result(self, addr, conv):
+        _fut = self._addr_2_rudp_conned_fut.get(addr, None)
+        if _fut is None:
+            return
+        try:
+            _fut.set_result(conv)
+        except asyncio.InvalidStateError:
+            # 并发情况即使`fut`已经done了也无所谓, 不处理即可
+            pass
+
     def set_is_proxy(self, is_proxy):
         self._is_proxy = is_proxy
 
-    def get_conn(self, addr, conn_type=None) -> ConnBase:
-        if conn_type is not None:
-            return self._conn_type_2_addr_2_conn[conn_type].get(addr, None)
-        _conn = self._conn_type_2_addr_2_conn[CONN_TYPE_RUDP].get(addr, None)
+    def get_conn(self, addr, proto_type=None) -> ConnBase:
+        if proto_type is not None:
+            return self._proto_type_2_addr_2_conn[proto_type].get(addr, None)
+        _conn = self._proto_type_2_addr_2_conn[PROTO_TYPE_RUDP].get(addr, None)
         if _conn is None:
-            _conn = self._conn_type_2_addr_2_conn[CONN_TYPE_TCP].get(addr, None)
+            _conn = self._proto_type_2_addr_2_conn[PROTO_TYPE_TCP].get(addr, None)
         return _conn
 
     def add_incoming_conn(
-            self, conn_type, transport: transports.BaseTransport,
+            self, proto_type, transport: transports.BaseTransport,
             peer_addr: Optional[Tuple[str, int]] = None,
             rudp_conv: int = None
             # rpc_handler: Optional[RpcHandler] = None
     ) -> ConnBase:
-        if conn_type == CONN_TYPE_TCP:
+        if proto_type == PROTO_TYPE_TCP:
             from TcpConn import TcpConn
             _conn = TcpConn(
                 ROLE_TYPE_PASSIVE, peer_addr,
-                close_cb=lambda ct=conn_type, a=peer_addr: self._remove_conn(ct, a),
+                close_cb=lambda ct=proto_type, a=peer_addr: self._remove_conn(ct, a),
                 is_proxy=self._is_proxy,
                 transport=transport)
         else:
             from RudpConn import RudpConn
             assert rudp_conv > 0
             _conn = RudpConn(
                 ROLE_TYPE_PASSIVE, peer_addr,
-                close_cb=lambda ct=conn_type, a=peer_addr: self._remove_conn(ct, a),
+                close_cb=lambda ct=proto_type, a=peer_addr: self._remove_conn(ct, a),
                 is_proxy=self._is_proxy,
                 transport=transport,
                 conv=rudp_conv)
 
-        self._conn_type_2_addr_2_conn[conn_type][peer_addr] = _conn
+        self._proto_type_2_addr_2_conn[proto_type][peer_addr] = _conn
         return _conn
 
     async def open_conn_by_addr(
-            self, conn_type, addr: typing.Tuple[str, int],
+            self, proto_type, addr: typing.Tuple[str, int],
             rpc_handler: RpcHandler = None
     ) -> (ConnBase, bool):
 
-        _conn = self._conn_type_2_addr_2_conn[conn_type].get(addr, None)
+        _conn = self._proto_type_2_addr_2_conn[proto_type].get(addr, None)
         is_conned = True
         if _conn is None:
-            if conn_type == CONN_TYPE_TCP:
+            if proto_type == PROTO_TYPE_TCP:
                 from TcpConn import TcpConn
                 conn_cls = TcpConn
                 # addr = transport.get_extra_info('peername')  # type: typing.Tuple[str, int]
                 # _conn = TcpConn.TcpConn(
                 #     ROLE_TYPE_ACTIVE, addr,
-                #     close_cb=lambda ct=conn_type, a=addr: self._remove_conn(ct, a),
+                #     close_cb=lambda ct=proto_type, a=addr: self._remove_conn(ct, a),
                 #     is_proxy=self._is_proxy,
                 #     # transport=transport
                 # )
@@ -221,19 +231,19 @@ async def open_conn_by_addr(
             _conn = conn_cls(
                 ROLE_TYPE_ACTIVE, addr,
                 rpc_handler=rpc_handler,
-                close_cb=lambda ct=conn_type, a=addr: self._remove_conn(ct, a),
+                close_cb=lambda ct=proto_type, a=addr: self._remove_conn(ct, a),
                 is_proxy=self._is_proxy,
                 # transport=transport
             )
 
-            self._conn_type_2_addr_2_conn[conn_type][addr] = _conn
+            self._proto_type_2_addr_2_conn[proto_type][addr] = _conn
             # if rpc_handler is not None:
             #     _conn.add_rpc_handler(rpc_handler)
             is_conned = await _conn.try_connect()
         return _conn, is_conned
 
-    def _remove_conn(self, conn_type, addr: typing.Tuple[str, int]):
-        self._conn_type_2_addr_2_conn[conn_type].pop(addr, None)
+    def _remove_conn(self, proto_type, addr: typing.Tuple[str, int]):
+        self._proto_type_2_addr_2_conn[proto_type].pop(addr, None)
 
     # def add_incoming_conn(
     #         self,

pycharm2020.1.3/script/ConnMgr.py

@@ -13,7 +13,7 @@
 
 import core.util.UtilApi
 from ConnBase import ConnBase
-from ConnMgr import ConnMgr, CONN_TYPE_TCP, CONN_TYPE_RUDP
+from ConnMgr import ConnMgr, PROTO_TYPE_TCP, PROTO_TYPE_RUDP
 from core.common.IdManager import IdManager
 from core.mobilelog.LogManager import AsyncLogger
 from core.util.TimerHub import TimerHub
@@ -60,11 +60,15 @@ class RpcHandler:
 
     def __init__(
             self, rpc_handler_id: bytes, conn: ConnBase = None,
-            entity: ServerEntity = None):
+            entity: ServerEntity = None, conn_proto_type: int = PROTO_TYPE_TCP):
         self.rpc_handler_id = rpc_handler_id
         self._logger = LogManager.get_logger()  # type: AsyncLogger
         # self._logger.info(f'!!! xx {rpc_handler_id=}')
-        self._conn = conn  # type: TcpConn
+        self._conn = conn  # type: ConnBase
+        if conn:
+            self._conn_proto_type = conn.get_proto_type()
+        else:
+            self._conn_proto_type = conn_proto_type
         self._entity = entity  # type: ServerEntity
         self._next_reply_id = 0
         self._pending_requests = {}  # type: typing.Dict[int, RpcReplyFuture]
@@ -210,8 +214,9 @@ async def _handle_create_conn(self, addr: typing.Tuple[str, int] = None):
                     return
                 addr = self._conn.get_addr()
             # self._try_connect_times += 1
-            self._conn, is_conned = await ConnMgr.instance().open_conn_by_addr(CONN_TYPE_TCP, addr, self)
-            # self._conn, is_conned = await ConnMgr.instance().open_conn_by_addr(CONN_TYPE_RUDP, addr, self)
+            # self._conn, is_conned = await ConnMgr.instance().open_conn_by_addr(CONN_TYPE_TCP, addr, self)
+            self._conn, is_conned = await ConnMgr.instance().open_conn_by_addr(
+                self._conn_proto_type, addr, self)
             if not is_conned:
                 for _msg in self._msg_buffer:
                     if _msg[0] == RPC_TYPE_REQUEST:

pycharm2020.1.3/script/RpcHandler.py

@@ -9,7 +9,7 @@
 
 from ConnBase import HEARTBEAT_TIMEOUT, HEARTBEAT_INTERVAL, ConnBase, RECONNECT_MAX_TIMES, RECONNECT_INTERVAL, \
     CONN_STATE_CONNECTING, CONN_STATE_CONNECTED, STRUCT_PACK_FORMAT
-from ConnMgr import CONN_TYPE_TCP, CONN_TYPE_RUDP, RudpProtocol, RUDP_HANDSHAKE_SYN, ConnMgr, ROLE_TYPE_PASSIVE
+from ConnMgr import PROTO_TYPE_TCP, PROTO_TYPE_RUDP, RudpProtocol, RUDP_HANDSHAKE_SYN, ConnMgr, ROLE_TYPE_PASSIVE
 from common import gv
 from core.common import rudp
 from core.common.IdManager import IdManager
@@ -39,9 +39,8 @@ def __init__(
             transport: transports.BaseTransport = None,
             conv: int = None,
     ):
-        self._conn_type = CONN_TYPE_RUDP
-
         super(RudpConn, self).__init__(role_type, addr, rpc_handler, close_cb, is_proxy, transport)
+        self._proto_type = PROTO_TYPE_RUDP
         self._conv = conv
         if self._role_type == ROLE_TYPE_PASSIVE:
             self._init_kcp()
@@ -53,14 +52,12 @@ def _init_kcp(self):
 
     async def try_connect(self) -> bool:
         self._conn_state = CONN_STATE_CONNECTING
-
         while self._try_connect_times < RECONNECT_MAX_TIMES:
             self._try_connect_times += 1
-            transport, protocol = await gv.get_ev_loop().create_datagram_endpoint(
+            self._transport, protocol = await gv.get_ev_loop().create_datagram_endpoint(
                 lambda: RudpProtocol(),
                 remote_addr=self._addr)
             self._transport.sendto(RUDP_HANDSHAKE_SYN)
-            self._transport = transport
             fut = ConnMgr.instance().create_rudp_conned_fut(self._addr)
             # self._logger.error(str(e))
             # await asyncio.sleep(RECONNECT_INTERVAL)
@@ -69,7 +66,7 @@ async def try_connect(self) -> bool:
                 conv = await asyncio.wait_for(
                     asyncio.shield(fut), timeout=RECONNECT_INTERVAL)
             except asyncio.exceptions.TimeoutError:
-                pass
+                self._logger.warning(f"try reconnect rudp: {str(self._addr)} ... {self._try_connect_times}")
             else:
                 self._conv = conv
                 self.set_connection_state(CONN_STATE_CONNECTED)
@@ -78,7 +75,7 @@ async def try_connect(self) -> bool:
                 return True
         else:
             pass  # todo
-            self._logger.error(f"try {RECONNECT_MAX_TIMES} times , still can't connect remote addr: {addr}")
+            self._logger.error(f"try {RECONNECT_MAX_TIMES} times , still can't connect rudp remote addr: {self._addr}")
             self._try_connect_times = 0
             return False
 

pycharm2020.1.3/script/RudpConn.py

@@ -18,9 +18,11 @@
 import typing
 
 # from core.util.performance.cpu_load_handler import CpuLoad
+from sanic_jwt_extended.jwt_manager import JWT
+
 import ConnBase
 # from ConnBase import ROLE_TYPE_PASSIVE
-from ConnMgr import ConnMgr, CONN_TYPE_RUDP, RudpProtocol
+from ConnMgr import ConnMgr, PROTO_TYPE_RUDP, RudpProtocol
 from ProxyRpcHandler import ProxyCliRpcHandler
 from core.util import UtilApi
 from core.util.UtilApi import wait_or_not, Singleton

pycharm2020.1.3/script/RudpServer.py

@@ -9,7 +9,7 @@
 
 from ConnBase import HEARTBEAT_TIMEOUT, HEARTBEAT_INTERVAL, ConnBase, RPC_HANDLER_ID_LEN, RECONNECT_MAX_TIMES, \
     RECONNECT_INTERVAL, CONN_STATE_CONNECTED, CONN_STATE_CONNECTING
-from ConnMgr import CONN_TYPE_TCP, ROLE_TYPE_ACTIVE
+from ConnMgr import PROTO_TYPE_TCP, ROLE_TYPE_ACTIVE
 from common import gv
 from core.common.IdManager import IdManager
 from core.util.TimerHub import TimerHub
@@ -38,7 +38,7 @@ def __init__(
             transport: transports.BaseTransport = None
     ):
         super(TcpConn, self).__init__(role_type, addr, rpc_handler, close_cb, is_proxy, transport)
-        self._conn_type = CONN_TYPE_TCP
+        self._proto_type = PROTO_TYPE_TCP
 
     # @wait_or_not()
     async def try_connect(self) -> bool:
@@ -62,7 +62,7 @@ async def try_connect(self) -> bool:
                 return True
         else:
             pass  # todo
-            self._logger.error(f"try {RECONNECT_MAX_TIMES} times , still can't connect remote addr: {addr}")
+            self._logger.error(f"try {RECONNECT_MAX_TIMES} times , still can't connect tcp remote addr: {self._addr}")
             self._try_connect_times = 0
             return False
 

pycharm2020.1.3/script/TcpConn.py

@@ -21,7 +21,7 @@
 
 import ConnBase
 # from ConnBase import ROLE_TYPE_PASSIVE, ROLE_TYPE_ACTIVE
-from ConnMgr import ConnMgr, CONN_TYPE_TCP, TcpProtocol, ROLE_TYPE_PASSIVE
+from ConnMgr import ConnMgr, PROTO_TYPE_TCP, TcpProtocol, ROLE_TYPE_PASSIVE
 from ProxyRpcHandler import ProxyCliRpcHandler
 from ServerBase import ServerBase
 from core.util import UtilApi