56 lines
1.5 KiB
Python
56 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
from channels.generic.websocket import WebsocketConsumer
|
|
|
|
from django.contrib.auth.models import User
|
|
|
|
import json
|
|
|
|
from .objects.GameManager import GameManager
|
|
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from .objects.Spectator import Spectator
|
|
from .objects.Player import Player
|
|
from .objects.Game import Game
|
|
|
|
game_manager: GameManager = GameManager()
|
|
|
|
class GameWebSocket(WebsocketConsumer):
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.channel_name = "games"
|
|
self.group_name = "games"
|
|
|
|
def connect(self):
|
|
|
|
self.user: User = self.scope["user"]
|
|
if (self.user.pk is None):
|
|
self.user.pk = 0
|
|
|
|
self.channel_layer.group_add(self.group_name, self.channel_name)
|
|
|
|
self.game_id = int(self.scope['url_route']['kwargs']['game_id'])
|
|
|
|
self.game: Game = game_manager.get(self.game_id)
|
|
|
|
if (self.game is None):
|
|
self.send("Game not found.")
|
|
self.disconnect(1017)
|
|
|
|
self.member: Player | Spectator = self.game.join(self.user.pk, self)
|
|
|
|
def receive(self, text_data: str = None, bytes_data: bytes = None):
|
|
|
|
if (text_data is None):
|
|
return
|
|
|
|
data: dict = json.loads(text_data)
|
|
|
|
self.member.receive(data)
|
|
|
|
def disconnect(self, close_code):
|
|
if (self.user.pk in self.game.players_id):
|
|
self.game.remove(self.member) |