|
|
- from dataclasses import dataclass, field
- from datetime import datetime
- from typing import Dict, Optional
-
- from paircd.client import Client
- from paircd.channel import Channel
- from paircd.reply import QUIT
-
-
- @dataclass
- class Server:
- clients_by_nick: Dict[str, Client] = field(default_factory=dict)
- channels_by_name: Dict[str, Channel] = field(default_factory=dict)
- create_time: datetime = field(default_factory=datetime.utcnow)
-
- def add_client(self, client: Client) -> None:
- self.clients_by_nick[client.nickname] = client
-
- def get_channel_by_name(self, name: str, create: bool = True) -> Optional[Channel]:
- if name not in self.channels_by_name and create:
- self.channels_by_name[name] = Channel(name=name)
- return self.channels_by_name.get(name)
-
- def get_client_by_name(self, name: str) -> Optional[Client]:
- return self.clients_by_nick.get(name)
-
- def remove_client_by_name(self, name: str) -> None:
- del self.clients_by_nick[name]
-
- def rename_client(self, name: str, new_name: str) -> None:
- if name not in self.clients_by_nick:
- raise KeyError(f"No client on server with nick {name}")
- if new_name in self.clients_by_nick:
- raise KeyError(f"Client already exists with nick {new_name}")
- self.clients_by_nick[new_name] = self.clients_by_nick[name]
- del self.clients_by_nick[name]
-
- async def disconnect_client(self, nick: str, quit_msg: str) -> None:
- client = self.clients_by_nick[nick]
-
- if client.closed:
- return
-
- client.log(f"is quitting: {quit_msg}")
- msg = QUIT(quit_msg, prefix=client.id())
- for client in self.clients_by_nick.values():
- if client.nickname == nick:
- continue
- client.write_message(msg)
-
- for channel_name in client.channels:
- channel = self.get_channel_by_name(channel_name, create=False)
- if channel is None:
- continue
- channel.remove_client_by_nick(client.nickname)
- if channel.get_channel_size() == 0:
- del self.channels_by_name[channel_name]
-
- self.remove_client_by_name(client.nickname)
- await client.close()
|