World logic

loon/mqtt/handlers/world.py

@@ -5,7 +5,7 @@ import redis
 
 from config import settings
 from loon.redis.chunk import set_chunk_cache
-from loon.web.users.state import user_threads
+from loon.web.users.state import user_threads, user_world_requests
 
 
 async def world_handler(client, userdata, msg):
@@ -21,14 +21,19 @@ async def world_handler(client, userdata, msg):
     except json.JSONDecodeError:
         return
 
-    x, z = match.group(1), match.group(2)
+    x, z = int(match.group(1)), int(match.group(2))
     envelope = json.dumps({"topic": f"world/chunk/{x}/{z}", "payload": data})
+    coord = (x, z)
 
     if settings.redis.enabled:
         try:
-            set_chunk_cache(int(x), int(z), data)
+            set_chunk_cache(x, z, data)
         except redis.RedisError:
             pass
 
-    for queue in list(user_threads.values()):
+    for uuid, queue in list(user_threads.items()):
+        wanted = user_world_requests.get(uuid)
+        if not wanted or coord not in wanted:
+            continue
         await queue.put(envelope)
+        wanted.discard(coord)

loon/web/users/router.py

@@ -6,7 +6,7 @@ from starlette.websockets import WebSocket, WebSocketDisconnect
 
 from loon.web import get_mqtt_manager
 from loon.web.auth.middleware import authenticated
-from loon.web.users.state import user_threads
+from loon.web.users.state import user_threads, user_world_requests
 
 router = APIRouter(prefix="/users", tags=["users"])
 
@@ -49,6 +49,7 @@ async def websocket_endpoint(websocket: WebSocket):
 
     queue = asyncio.Queue()
     user_threads[user.uuid] = queue
+    user_world_requests[user.uuid] = set()
 
     try:
         while True:
@@ -58,3 +59,4 @@ async def websocket_endpoint(websocket: WebSocket):
         pass
     finally:
         user_threads.pop(user.uuid, None)
+        user_world_requests.pop(user.uuid, None)

loon/web/users/state.py

@@ -1 +1,2 @@
 user_threads = {}
+user_world_requests = {}

loon/web/world/router.py

@@ -7,7 +7,7 @@ from config import settings
 from loon.redis.chunk import get_chunk_cache, reset_chunk_cache
 from loon.web import get_mqtt_manager
 from loon.web.auth.middleware import authenticated
-from loon.web.users.state import user_threads
+from loon.web.users.state import user_threads, user_world_requests
 
 router = APIRouter(prefix="/world", tags=["world"])
 
@@ -21,12 +21,20 @@ async def request_world(
     z_start: int,
     z_end: int,
 ):
+    uuid = request.user.user.uuid
+    wanted = user_world_requests.get(uuid)
+
     if not settings.redis.enabled:
+        if wanted is not None:
+            for x in range(x_start, x_end + 1):
+                for z in range(z_start, z_end + 1):
+                    wanted.update((x, z))
+
         get_mqtt_manager().publish(f"loon/world/chunks/{x_start}:{x_end}/{z_start}:{z_end}/request")
         return 200
 
     misses = []
-    thread = user_threads.get(request.user.user.uuid)
+    thread = user_threads.get(uuid)
 
     try:
         for x in range(x_start, x_end + 1):
@@ -39,8 +47,15 @@ async def request_world(
                             json.dumps({"topic": f"world/chunk/{x}/{z}", "payload": cached_chunk})
                         )
                 else:
+                    if wanted is not None:
+                        wanted.add((x, z))
                     misses.append((x, z))
-    except redis.RedisError as e:
+    except redis.RedisError:
+        if wanted is not None:
+            for x in range(x_start, x_end + 1):
+                for z in range(z_start, z_end + 1):
+                    wanted.update((x, z))
+
         get_mqtt_manager().publish(f"loon/world/chunks/{x_start}:{x_end}/{z_start}:{z_end}/request")
         return 200