Upgraded to Python 3.7 and using asyncio for coroutines now. Also made the client go up to Awaiting Snapshot.

dpnet/netserver.py

@@ -16,20 +16,28 @@ from giants.player import Player, PlayerPhases
 import time
 from giants import APPLICATION_GUID
 from .DN_MSG_INTERNAL_SEND_CONNECT_INFO import DN_MSG_INTERNAl_SEND_CONNECT_INFO
+import asyncio
 
 logger = setup_logger(__name__)
 
 
-class Netserver:
+class Netserver(asyncio.DatagramProtocol):
     def __init__(self, server):
-        self.socket = None
+        self.remotesocket = None
         self.statssocket = None
         self.server = server
         self.addrs = []
         self.guid = uuid.uuid4().bytes
         self.incoming_packets = []
+        super().__init__()
+
+    def connection_made(self, transport):
+        logger.debug("Connection made")
+        self.remotesocket = transport
 
     def run(self):
+        logger.debug("Run")
+        return
         self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
         self.socket.bind((self.server.listen_ip, self.server.listen_port))
         logger.debug("Listening to %s:%s", self.server.listen_ip, self.server.listen_port)
@@ -57,26 +65,25 @@ class Netserver:
     def send_packets(self):
         pass
 
-    def handle_packets(self):
+    def datagram_received(self, data, addr):
-        while True:
-            data, addr = self.socket.recvfrom(1024)
         logger.debug("%s:%s > %s", addr[0], addr[1], data.hex())
-            self.handle_new_packet(addr, data)
+        loop = asyncio.get_event_loop()
+        loop.create_task(self.handle_new_packet(addr, data))
 
-    def handle_new_packet(self, addr, bData):
+    async def handle_new_packet(self, addr, bData):
         pPacket = Packet(bData)
         first = pPacket.getByte()
         if len(bData) >= 4 and first & DFrame.PACKET_COMMAND_DATA:
-            self.handle_new_dframe(addr, pPacket)
+            await self.handle_new_dframe(addr, pPacket)
         elif len(bData) >= 12 and first & CFrame.PACKET_COMMAND_FRAME:
-            self.handle_new_cframe(addr, pPacket)
+            await self.handle_new_cframe(addr, pPacket)
         elif first == EnumQuery.LEAD or first == EnumResponse.LEAD:
-            self.handle_new_enum(addr, pPacket)
+            await self.handle_new_enum(addr, pPacket)
         else:
             logger.error("Unknown frame received")
             return
 
-    def handle_new_enum(self, addr, data):
+    async def handle_new_enum(self, addr, data):
         # EnumQuery or EnumResponse
         second = data.getByte()
         if second == EnumQuery.COMMAND:
@@ -134,7 +141,7 @@ class Netserver:
         else:
             logger.error("Unknown DPLHP command: %s", second)
 
-    def handle_new_cframe(self, addr, data):
+    async def handle_new_cframe(self, addr, data):
         try:
             cframe = CFrame(data)
             session = self.get_session(addr)
@@ -155,10 +162,11 @@ class Netserver:
                     return
                 else:
                     logger.debug("New connection. Sending a CFRAME CONNECTED.")
-                    session = Session(self.socket)
+                    session = Session(self.remotesocket)
                     session.SessID = cframe.SessID
                     session.ip = addr[0]
                     session.port = addr[1]
+                    await session.lock.acquire()
                     self.addrs.append(session)
                     session.send_cframe_connected(cframe)
                     player = Player("Unknown", session)
@@ -207,6 +215,9 @@ class Netserver:
                         session.send_cframe_sack()
                         sack_sent = True
 
+                # release lock for new packet to be sent
+                session.lock.release()
+
                 # WIP: The bNSeq, bNRcv, optional selective acknowledgment (SACK), and optional send mask fields are then processed by using the standard rules in sections 3.1.5.2.1 through 3.1.5.2.4
                 # TODO: A successfully validated SACK packet SHOULD count as a valid receive and thus restart the KeepAlive timer
 
@@ -222,7 +233,7 @@ class Netserver:
             traceback.print_exc()
             return
 
-    def handle_new_dframe(self, addr, data):
+    async def handle_new_dframe(self, addr, data):
         try:
             dframe = DFrame(data)
             logger.debug("Received DFRAME")
@@ -258,7 +269,7 @@ class Netserver:
 
             session.next_expected_seq += 1
             if dframe.Payload:
-                self.handle_game_packet(session, dframe.Payload)
+                await self.handle_game_packet(session, dframe.Payload)
             else:
                 pass
         except Exception as e:
@@ -266,7 +277,7 @@ class Netserver:
             traceback.print_exc()
             return
 
-    def handle_game_packet(self, session, payload):
+    async def handle_game_packet(self, session, payload):
         player = self.get_temp_player(session)
 
         if payload[0] == 0xc1:
@@ -325,10 +336,10 @@ class Netserver:
                 return
 
             player.phase = PlayerPhases.DN_ACK_CONNECT_INFO
-            self.server.add_player(player)
+            await self.server.add_player(player)
             #player.session.send_cframe_sack()
 
-            player.session.send_gamedata(b'\x3c'+struct.pack("<L", player.id)+b"\x00")
+            await player.session.send_gamedata(b'\x3c'+struct.pack("<L", player.id)+b"\x00")
 
             p1 = b'\x01' + struct.pack("<B", len(self.server.players))
             for pplayer in self.server.players:
@@ -344,9 +355,9 @@ class Netserver:
                 p1 += struct.pack("<B", pplayer.team)
             p1 += b'\x00\x00'
 
-            player.session.send_gamedata(p1, acknow=False)
+            await player.session.send_gamedata(p1, acknow=False)
-            player.session.send_gamedata(b'\x3d\x00\x5b\x53\x65\x72\x76\x65\x72\x5d'+b"\x00"*25) # [SERVER]
+            await player.session.send_gamedata(b'\x3d\x00\x5b\x53\x65\x72\x76\x65\x72\x5d'+b"\x00"*25) # [SERVER]
-            player.session.send_gamedata(b'\x3d\x01'+player.name.encode("ascii")+b"\x00"*(33-len(player.name.encode("ascii")))) # playername
+            await player.session.send_gamedata(b'\x3d\x01'+player.name.encode("ascii")+b"\x00"*(33-len(player.name.encode("ascii")))) # playername
             """
             player.session.send_gamedata(b'\x0f\x56\xab\x31\x96\x06\x00\x00\x00\x00\x00\x00\x00\x00')  # unknown
             player.session.send_gamedata(b'\x10\x02\x56\xab\x31\x96\x00\x00\x00\x00\x00\x00\x00\x00')  # unknown
@@ -368,10 +379,12 @@ class Netserver:
             player.session.send_gamedata(b'\x0a\x12\x00\x00\x00\x00\x2e\x10\x03\x10\x12\x00\x00\x00\xa4\xfd\x2c\x03\x00',acknow=True)  # unknown
             player.session.send_gamedata(b'\x2f\x02\x00\x00\xc2\x00\x00\x00\x00\x00\x00\x00\x00\x00', acknow=True)  # unknown
             """
+        else:
+            logger.debug("OPCODE: %s VALUES: %s", payload[0], payload[1:])
 
 
     def send_packet(self, addr, packet):
-        self.socket.sendto(packet.getvalue(), addr)
+        self.remotesocket.sendto(packet.getvalue(), addr)
         logger.debug("%s:%s < %s", addr[0], addr[1], packet.getvalue().hex())
 
     def get_session(self, iporaddr, port=None):

dpnet/session.py

@@ -2,6 +2,7 @@ from threading import Timer
 from utils.logger import setup_logger
 from dpnet.CFrame import CFrame
 from dpnet.DFrame import DFrame
+import asyncio
 
 logger = setup_logger(__name__)
 
@@ -18,6 +19,7 @@ class Session:
         self.LastMsgID = 0
         self.connect_retry_timer_num = 0
         self.connect_retry_timer = False
+        self.lock = asyncio.Lock()
 
     def send(self, packet):
         rawpacket = packet.to_packet()
@@ -49,7 +51,7 @@ class Session:
         self.send(response)
         self.connect_retry_timer = self.setup_Connect_Retry_Timer()
 
-    def send_gamedata(self, bPayload, **kwargs):
+    async def send_gamedata(self, bPayload, **kwargs):
         acknow = kwargs.get("acknow", True)
         dframe = DFrame()
         dframe.Command = DFrame.PACKET_COMMAND_DATA | DFrame.PACKET_COMMAND_RELIABLE | DFrame.PACKET_COMMAND_SEQUENTIAL | DFrame.PACKET_COMMAND_NEW_MSG | DFrame.PACKET_COMMAND_END_MSG
@@ -60,8 +62,8 @@ class Session:
         dframe.NRcv = self.next_expected_seq
         dframe.Payload = bPayload
         self.send(dframe)
-        # lock acquire
+        if acknow:
-        # lock release
+            await self.lock.acquire()
         return dframe
 
     def send_cframe_connected(self, connect):

plugins/greetings.py

@@ -2,11 +2,11 @@ class Greetings:
     def __init__(self, server):
         self.server = server
 
-    def on_player_join(self, player):
+    async def on_player_join(self, player):
-        self.server.broadcast_message("Welcome "+player.name+"!")
+        await self.server.broadcast_message("Welcome "+player.name+"!")
 
-    def on_map_change(self, newmap):
+    async def on_map_change(self, newmap):
-        self.server.broadcast_message("You are now playing on "+newmap.mapname)
+        await self.server.broadcast_message("You are now playing on "+newmap.mapname)
 
 
 def setup(server):

server.py

@@ -8,6 +8,7 @@ import os
 from giants import Teams, GameTypes
 from utils.logger import setup_logger
 import traceback
+import asyncio
 
 logger = setup_logger(__name__)
 
@@ -42,42 +43,42 @@ class Server:
         #logger.debug("Calling update")
         pass
 
-    def add_player(self, player):
+    async def add_player(self, player):
         self.tempplayers.remove(player)
         self.players.append(player)
-        self._broadcast_event("on_player_join", player)
+        await self._broadcast_event("on_player_join", player)
 
     # todo: remove
     def create_temp_player(self, player):
         self.tempplayers.append(player)
 
-    def remove_player(self, player):
+    async def remove_player(self, player):
         self.players.remove(player)
-        self._broadcast_event("on_player_left", player)
+        await self._broadcast_event("on_player_left", player)
 
     def add_plugin(self, plugin):
         self._plugins.append(plugin)
 
-    def change_map(self, mappath):
+    async def change_map(self, mappath):
         try:
             self.currentmap = Map(mappath)
-            self._broadcast_event("on_map_change", self.currentmap)
+            await self._broadcast_event("on_map_change", self.currentmap)
         except Exception:
             logger.error("Could not change map")
 
-    def _broadcast_event(self, event, *args):
+    async def _broadcast_event(self, event, *args):
         logger.debug("Broadcasting event "+event)
         for plugin in self._plugins:
             if hasattr(plugin, event):
                 func = getattr(plugin, event)
                 if callable(func):
                     try:
-                        func(*args)
+                        await func(*args)
                     except Exception:
                         logger.error("Could not call plugin function: "+plugin.__class__.__name__+"."+event)
                         traceback.print_exc()
 
-    def _load_plugins(self):
+    def load_plugins(self):
         plugins = os.listdir("plugins")
         for plugin in plugins:
             if os.path.isdir("plugins/"+plugin) or plugin == "__init__.py":
@@ -93,18 +94,23 @@ class Server:
                         logger.info("Loading plugin "+module.__name__)
                         module.setup(self)
             except Exception:
-                logger.warn("Could not load plugin "+plugin)
+                logger.warning("Could not load plugin "+plugin)
                 traceback.print_exc()
 
-    def broadcast_message(self, text):
+    async def broadcast_message(self, text):
         for player in self.players:
-            player.session.send_gamedata(b'\x35\x80\x81'+text.encode("ascii")+b'\x00\x00')
+            await player.session.send_gamedata(b'\x35\x80\x81'+text.encode("ascii")+b'\x00\x00')
-
-    def run(self):
-        self._load_plugins()
-        return Netserver(self).run()
-
 
 if __name__ == '__main__':
-    server = Server(name="giantsd", maxplayers=400, register=False)
+    server = Server(name="giantsd", maxplayers=20, register=False)
-    server.run() # blocking
+    server.load_plugins()
+
+    loop = asyncio.get_event_loop()
+    listen = loop.create_datagram_endpoint(lambda: Netserver(server), local_addr=(server.listen_ip, server.listen_port))
+    transport, protocol = loop.run_until_complete(listen)
+    try:
+        loop.run_forever()
+    except KeyboardInterrupt:
+        pass
+    transport.close()
+    loop.close()