You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
60 lines
2.2 KiB
60 lines
2.2 KiB
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()
|