Ross
228 lines
8.3 KiB
Python
228 lines
8.3 KiB
Python
import logging
|
|
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__)
|
|
|
|
class MultiuserConsumer(AsyncJsonWebsocketConsumer):
|
|
@database_sync_to_async
|
|
def get_session(self, session_id):
|
|
try:
|
|
return MultiuserSession.objects.select_related('creator').get(id=session_id)
|
|
except (MultiuserSession.DoesNotExist, ValidationError):
|
|
return None
|
|
|
|
async def connect(self):
|
|
self.room_name = self.scope['url_route']['kwargs']['room_name']
|
|
self.room_group_name = f'multiuser_{self.room_name}'
|
|
self.user = self.scope.get('user')
|
|
|
|
# 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
|
|
await self.close()
|
|
return
|
|
|
|
# Determine username and staff privilege
|
|
if self.user and self.user.is_authenticated:
|
|
self.username = self.user.username
|
|
# Manager is the creator or a site admin/staff
|
|
self.is_staff = (self.user == session.creator) or self.user.is_staff
|
|
else:
|
|
self.username = f"Guest_{self.channel_name[-4:]}"
|
|
self.is_staff = False
|
|
|
|
# Add to channels group
|
|
await self.channel_layer.group_add(
|
|
self.room_group_name,
|
|
self.channel_name
|
|
)
|
|
await self.accept()
|
|
await self.send_json({
|
|
'type': 'welcome',
|
|
'channel_name': self.channel_name
|
|
})
|
|
|
|
# Update room state in cache & notify others
|
|
await self.add_user_to_room()
|
|
await self.broadcast_room_status()
|
|
|
|
async def disconnect(self, close_code):
|
|
# Remove from channels group
|
|
await self.channel_layer.group_discard(
|
|
self.room_group_name,
|
|
self.channel_name
|
|
)
|
|
|
|
# Update room state in cache & notify others
|
|
await self.remove_user_from_room()
|
|
await self.broadcast_room_status()
|
|
|
|
async def receive_json(self, content):
|
|
msg_type = content.get('type')
|
|
|
|
if msg_type == 'viewer_state':
|
|
# Replicate view state changes from the active controller to others
|
|
if await self.has_control():
|
|
await self.channel_layer.group_send(
|
|
self.room_group_name,
|
|
{
|
|
'type': 'viewer_state_update',
|
|
'sender': self.channel_name,
|
|
'data': content.get('data')
|
|
}
|
|
)
|
|
elif msg_type == 'annotations':
|
|
# Replicate annotation changes from the active controller to others
|
|
if await self.has_control():
|
|
await self.channel_layer.group_send(
|
|
self.room_group_name,
|
|
{
|
|
'type': 'annotations_update',
|
|
'sender': self.channel_name,
|
|
'data': content.get('data')
|
|
}
|
|
)
|
|
elif msg_type == 'request_control':
|
|
# Broadcast a control request to all participants (so the admin/controller can grant it)
|
|
await self.channel_layer.group_send(
|
|
self.room_group_name,
|
|
{
|
|
'type': 'control_request',
|
|
'sender_name': self.username,
|
|
'sender_channel': self.channel_name
|
|
}
|
|
)
|
|
elif msg_type == 'take_control':
|
|
# Supervisors/admins can immediately assume control
|
|
if self.is_staff:
|
|
await self.set_controller(self.channel_name, self.username)
|
|
await self.broadcast_room_status()
|
|
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:
|
|
await self.set_controller(target_channel, target_username)
|
|
await self.broadcast_room_status()
|
|
elif msg_type == 'release_control':
|
|
# Relinquish control
|
|
if await self.has_control():
|
|
await self.set_controller(None, None)
|
|
await self.broadcast_room_status()
|
|
|
|
# Channels group event handlers
|
|
async def viewer_state_update(self, event):
|
|
if event['sender'] != self.channel_name:
|
|
await self.send_json({
|
|
'type': 'viewer_state',
|
|
'data': event['data']
|
|
})
|
|
|
|
async def annotations_update(self, event):
|
|
if event['sender'] != self.channel_name:
|
|
await self.send_json({
|
|
'type': 'annotations',
|
|
'data': event['data']
|
|
})
|
|
|
|
async def control_request(self, event):
|
|
await self.send_json({
|
|
'type': 'control_request',
|
|
'sender_name': event['sender_name'],
|
|
'sender_channel': event['sender_channel']
|
|
})
|
|
|
|
async def room_status_update(self, event):
|
|
await self.send_json({
|
|
'type': 'room_status',
|
|
'users': event['users'],
|
|
'controller_channel': event['controller_channel'],
|
|
'controller_username': event['controller_username']
|
|
})
|
|
|
|
async def change_case(self, event):
|
|
await self.send_json({
|
|
'type': 'change_case',
|
|
'case_id': event['case_id']
|
|
})
|
|
|
|
# Cache state helpers (using sync_to_async for thread-safe Memcached/Redis interactions)
|
|
def get_cache_key(self):
|
|
return f"multiuser_room_{self.room_name}"
|
|
|
|
@sync_to_async
|
|
def _get_state(self):
|
|
return cache.get(self.get_cache_key())
|
|
|
|
@sync_to_async
|
|
def _set_state(self, state):
|
|
cache.set(self.get_cache_key(), state, timeout=86400) # Expire after 1 day
|
|
|
|
async def add_user_to_room(self):
|
|
state = await self._get_state()
|
|
if not state:
|
|
state = {
|
|
'users': {},
|
|
'controller_channel': None,
|
|
'controller_username': 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']:
|
|
state['controller_channel'] = self.channel_name
|
|
state['controller_username'] = self.username
|
|
|
|
await self._set_state(state)
|
|
|
|
async def remove_user_from_room(self):
|
|
state = await self._get_state()
|
|
if state:
|
|
state['users'].pop(self.channel_name, None)
|
|
# If the leaving user was in control, elect a replacement
|
|
if state['controller_channel'] == self.channel_name:
|
|
if state['users']:
|
|
next_channel = list(state['users'].keys())[0]
|
|
state['controller_channel'] = next_channel
|
|
state['controller_username'] = state['users'][next_channel]['username']
|
|
else:
|
|
state['controller_channel'] = None
|
|
state['controller_username'] = None
|
|
|
|
await self._set_state(state)
|
|
|
|
async def has_control(self):
|
|
state = await self._get_state()
|
|
return state and state['controller_channel'] == self.channel_name
|
|
|
|
async def set_controller(self, channel_name, username):
|
|
state = await self._get_state()
|
|
if state:
|
|
state['controller_channel'] = channel_name
|
|
state['controller_username'] = username
|
|
await self._set_state(state)
|
|
|
|
async def broadcast_room_status(self):
|
|
state = await self._get_state()
|
|
if state:
|
|
await self.channel_layer.group_send(
|
|
self.room_group_name,
|
|
{
|
|
'type': 'room_status_update',
|
|
'users': state['users'],
|
|
'controller_channel': state['controller_channel'],
|
|
'controller_username': state['controller_username']
|
|
}
|
|
)
|