try to fix multiuser sync
This commit is contained in:
Ross
+99
-11
@@ -1,12 +1,11 @@
|
||||
import logging
|
||||
from loguru import logger
|
||||
from channels.generic.websocket import AsyncJsonWebsocketConsumer
|
||||
from django.core.cache import cache
|
||||
from asgiref.sync import sync_to_async
|
||||
from channels.db import database_sync_to_async
|
||||
from django.core.exceptions import ValidationError
|
||||
from atlas.models import MultiuserSession
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
import urllib.parse
|
||||
|
||||
class MultiuserConsumer(AsyncJsonWebsocketConsumer):
|
||||
@database_sync_to_async
|
||||
@@ -24,7 +23,7 @@ class MultiuserConsumer(AsyncJsonWebsocketConsumer):
|
||||
# Get session from DB using self.room_name
|
||||
session = await self.get_session(self.room_name)
|
||||
if not session:
|
||||
# Reject connection if session does not exist
|
||||
logger.warning("[Multiuser WS] Rejecting connection: session {} not found", self.room_name)
|
||||
await self.close()
|
||||
return
|
||||
|
||||
@@ -37,6 +36,16 @@ class MultiuserConsumer(AsyncJsonWebsocketConsumer):
|
||||
self.username = f"Guest_{self.channel_name[-4:]}"
|
||||
self.is_staff = False
|
||||
|
||||
# Parse query string for client_type (default to 'viewer')
|
||||
query_string = self.scope.get('query_string', b'').decode('utf-8')
|
||||
params = urllib.parse.parse_qs(query_string)
|
||||
self.client_type = params.get('client_type', ['viewer'])[0]
|
||||
|
||||
logger.info(
|
||||
"[Multiuser WS] Connect request: room={}, user={}, type={}, channel={}",
|
||||
self.room_name, self.username, self.client_type, self.channel_name
|
||||
)
|
||||
|
||||
# Add to channels group
|
||||
await self.channel_layer.group_add(
|
||||
self.room_group_name,
|
||||
@@ -48,11 +57,21 @@ class MultiuserConsumer(AsyncJsonWebsocketConsumer):
|
||||
'channel_name': self.channel_name
|
||||
})
|
||||
|
||||
# Update room state in cache & notify others
|
||||
await self.add_user_to_room()
|
||||
await self.broadcast_room_status()
|
||||
if self.client_type == 'viewer':
|
||||
# Update room state in cache & notify others
|
||||
await self.add_user_to_room()
|
||||
await self.broadcast_room_status()
|
||||
else:
|
||||
# Passive session listener: just broadcast the current room status to this new client
|
||||
logger.info("[Multiuser WS] Passive listener connected: {}", self.channel_name)
|
||||
await self.broadcast_room_status()
|
||||
|
||||
async def disconnect(self, close_code):
|
||||
logger.info(
|
||||
"[Multiuser WS] Disconnect: room={}, user={}, type={}, channel={}, code={}",
|
||||
self.room_name, self.username, self.client_type, self.channel_name, close_code
|
||||
)
|
||||
|
||||
# Remove from channels group
|
||||
await self.channel_layer.group_discard(
|
||||
self.room_group_name,
|
||||
@@ -60,15 +79,28 @@ class MultiuserConsumer(AsyncJsonWebsocketConsumer):
|
||||
)
|
||||
|
||||
# Update room state in cache & notify others
|
||||
await self.remove_user_from_room()
|
||||
await self.broadcast_room_status()
|
||||
if self.client_type == 'viewer':
|
||||
await self.remove_user_from_room()
|
||||
await self.broadcast_room_status()
|
||||
|
||||
async def receive_json(self, content):
|
||||
msg_type = content.get('type')
|
||||
logger.info(
|
||||
"[Multiuser WS] Received message from user={} (type={}): {}",
|
||||
self.username, msg_type, content
|
||||
)
|
||||
|
||||
if self.client_type != 'viewer':
|
||||
logger.warning(
|
||||
"[Multiuser WS] Rejected message from passive listener user={} (type={})",
|
||||
self.username, msg_type
|
||||
)
|
||||
return
|
||||
|
||||
if msg_type == 'viewer_state':
|
||||
# Replicate view state changes from the active controller to others
|
||||
if await self.has_control():
|
||||
logger.debug("[Multiuser WS] Broadcasting viewer_state from user={}", self.username)
|
||||
await self.channel_layer.group_send(
|
||||
self.room_group_name,
|
||||
{
|
||||
@@ -77,9 +109,15 @@ class MultiuserConsumer(AsyncJsonWebsocketConsumer):
|
||||
'data': content.get('data')
|
||||
}
|
||||
)
|
||||
else:
|
||||
logger.warning(
|
||||
"[Multiuser WS] viewer_state rejected: user={} does not have control",
|
||||
self.username
|
||||
)
|
||||
elif msg_type == 'annotations':
|
||||
# Replicate annotation changes from the active controller to others
|
||||
if await self.has_control():
|
||||
logger.debug("[Multiuser WS] Broadcasting annotations from user={}", self.username)
|
||||
await self.channel_layer.group_send(
|
||||
self.room_group_name,
|
||||
{
|
||||
@@ -88,8 +126,14 @@ class MultiuserConsumer(AsyncJsonWebsocketConsumer):
|
||||
'data': content.get('data')
|
||||
}
|
||||
)
|
||||
else:
|
||||
logger.warning(
|
||||
"[Multiuser WS] annotations rejected: user={} does not have control",
|
||||
self.username
|
||||
)
|
||||
elif msg_type == 'request_control':
|
||||
# Broadcast a control request to all participants (so the admin/controller can grant it)
|
||||
logger.info("[Multiuser WS] Control request from user={}", self.username)
|
||||
await self.channel_layer.group_send(
|
||||
self.room_group_name,
|
||||
{
|
||||
@@ -101,25 +145,47 @@ class MultiuserConsumer(AsyncJsonWebsocketConsumer):
|
||||
elif msg_type == 'take_control':
|
||||
# Supervisors/admins can immediately assume control
|
||||
if self.is_staff:
|
||||
logger.info("[Multiuser WS] Supervisor user={} taking control", self.username)
|
||||
await self.set_controller(self.channel_name, self.username)
|
||||
await self.broadcast_room_status()
|
||||
else:
|
||||
logger.warning(
|
||||
"[Multiuser WS] take_control rejected: user={} is not supervisor/staff",
|
||||
self.username
|
||||
)
|
||||
elif msg_type == 'grant_control':
|
||||
# The active controller or a supervisor can hand over control to someone else
|
||||
if self.is_staff or await self.has_control():
|
||||
target_channel = content.get('target_channel')
|
||||
target_username = content.get('target_username')
|
||||
if target_channel:
|
||||
logger.info(
|
||||
"[Multiuser WS] Control granted by user={} to target_user={} ({})",
|
||||
self.username, target_username, target_channel
|
||||
)
|
||||
await self.set_controller(target_channel, target_username)
|
||||
await self.broadcast_room_status()
|
||||
else:
|
||||
logger.warning(
|
||||
"[Multiuser WS] grant_control rejected: user={} is not controller/staff",
|
||||
self.username
|
||||
)
|
||||
elif msg_type == 'release_control':
|
||||
# Relinquish control
|
||||
if await self.has_control():
|
||||
logger.info("[Multiuser WS] Control released by user={}", self.username)
|
||||
await self.set_controller(None, None)
|
||||
await self.broadcast_room_status()
|
||||
else:
|
||||
logger.warning(
|
||||
"[Multiuser WS] release_control rejected: user={} is not active controller",
|
||||
self.username
|
||||
)
|
||||
|
||||
# Channels group event handlers
|
||||
async def viewer_state_update(self, event):
|
||||
if event['sender'] != self.channel_name:
|
||||
logger.debug("[Multiuser WS] Forwarding viewer_state_update to channel={}", self.channel_name)
|
||||
await self.send_json({
|
||||
'type': 'viewer_state',
|
||||
'data': event['data']
|
||||
@@ -127,12 +193,14 @@ class MultiuserConsumer(AsyncJsonWebsocketConsumer):
|
||||
|
||||
async def annotations_update(self, event):
|
||||
if event['sender'] != self.channel_name:
|
||||
logger.debug("[Multiuser WS] Forwarding annotations_update to channel={}", self.channel_name)
|
||||
await self.send_json({
|
||||
'type': 'annotations',
|
||||
'data': event['data']
|
||||
})
|
||||
|
||||
async def control_request(self, event):
|
||||
logger.debug("[Multiuser WS] Forwarding control_request to channel={}", self.channel_name)
|
||||
await self.send_json({
|
||||
'type': 'control_request',
|
||||
'sender_name': event['sender_name'],
|
||||
@@ -140,6 +208,7 @@ class MultiuserConsumer(AsyncJsonWebsocketConsumer):
|
||||
})
|
||||
|
||||
async def room_status_update(self, event):
|
||||
logger.debug("[Multiuser WS] Forwarding room_status_update to channel={}", self.channel_name)
|
||||
await self.send_json({
|
||||
'type': 'room_status',
|
||||
'users': event['users'],
|
||||
@@ -148,6 +217,7 @@ class MultiuserConsumer(AsyncJsonWebsocketConsumer):
|
||||
})
|
||||
|
||||
async def change_case(self, event):
|
||||
logger.info("[Multiuser WS] Forwarding change_case (case_id={}) to channel={}", event['case_id'], self.channel_name)
|
||||
await self.send_json({
|
||||
'type': 'change_case',
|
||||
'case_id': event['case_id']
|
||||
@@ -174,15 +244,28 @@ class MultiuserConsumer(AsyncJsonWebsocketConsumer):
|
||||
'controller_username': None
|
||||
}
|
||||
|
||||
# Remove any existing channel names for the same username to prevent duplicate presence counts on refresh
|
||||
for old_channel in list(state['users'].keys()):
|
||||
if state['users'][old_channel]['username'] == self.username:
|
||||
logger.info(
|
||||
"[Multiuser WS] Pruning stale channel name {} for user={}",
|
||||
old_channel, self.username
|
||||
)
|
||||
state['users'].pop(old_channel, None)
|
||||
|
||||
state['users'][self.channel_name] = {
|
||||
'username': self.username,
|
||||
'is_staff': self.is_staff
|
||||
}
|
||||
|
||||
# Auto-assign control if no active controller exists
|
||||
if not state['controller_channel']:
|
||||
# Auto-assign control if no active controller exists or the previous controller was pruned
|
||||
if not state['controller_channel'] or state['controller_channel'] not in state['users']:
|
||||
state['controller_channel'] = self.channel_name
|
||||
state['controller_username'] = self.username
|
||||
logger.info(
|
||||
"[Multiuser WS] Control assigned to user={} (channel={})",
|
||||
self.username, self.channel_name
|
||||
)
|
||||
|
||||
await self._set_state(state)
|
||||
|
||||
@@ -196,9 +279,14 @@ class MultiuserConsumer(AsyncJsonWebsocketConsumer):
|
||||
next_channel = list(state['users'].keys())[0]
|
||||
state['controller_channel'] = next_channel
|
||||
state['controller_username'] = state['users'][next_channel]['username']
|
||||
logger.info(
|
||||
"[Multiuser WS] Control passed to user={} (channel={})",
|
||||
state['controller_username'], next_channel
|
||||
)
|
||||
else:
|
||||
state['controller_channel'] = None
|
||||
state['controller_username'] = None
|
||||
logger.info("[Multiuser WS] Room is empty, control cleared")
|
||||
|
||||
await self._set_state(state)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user