Continuing the refactoring

This commit is contained in:
Rafael Vargas
2023-02-19 11:28:33 -03:00
parent 7a5d76ffd3
commit 72043c4475
26 changed files with 369 additions and 502 deletions

View File

@@ -1,23 +1,70 @@
import asyncio
from multiprocessing import Lock, Queue
from enum import Enum
from multiprocessing import Lock, Process, Queue
from multiprocessing.managers import BaseManager, NamespaceProxy
from queue import Empty
from threading import Thread
from typing import Dict, Tuple, Union
from Config.Singleton import Singleton
from discord import Guild, Interaction
from discord import Guild, Interaction, TextChannel
from discord.ext.commands import Context
from Parallelism.AbstractProcessManager import AbstractPlayersManager
from Parallelism.ProcessExecutor import ProcessCommandsExecutor
from Music.Song import Song
from Parallelism.PlayerProcess import PlayerProcess
from Music.Playlist import Playlist
from Parallelism.ProcessInfo import PlayerInfo, ProcessStatus
from Parallelism.Commands import VCommands, VCommandsType
from Music.VulkanBot import VulkanBot
class ProcessManager(Singleton, AbstractPlayersManager):
class ProcessStatus(Enum):
RUNNING = 'Running'
SLEEPING = 'Sleeping'
class PlayerProcessInfo:
"""
Class to store the reference to all structures to maintain a process player
"""
def __init__(self, process: Process, queueToPlayer: Queue, queueToMain: Queue, playlist: Playlist, lock: Lock, textChannel: TextChannel) -> None:
self.__process = process
self.__queueToPlayer = queueToPlayer
self.__queueToMain = queueToMain
self.__playlist = playlist
self.__lock = lock
self.__textChannel = textChannel
self.__status = ProcessStatus.RUNNING
def setProcess(self, newProcess: Process) -> None:
self.__process = newProcess
def getStatus(self) -> ProcessStatus:
return self.__status
def setStatus(self, status: ProcessStatus) -> None:
self.__status = status
def getProcess(self) -> Process:
return self.__process
def getQueueToPlayer(self) -> Queue:
return self.__queueToPlayer
def getQueueToMain(self) -> Queue:
return self.__queueToMain
def getPlaylist(self) -> Playlist:
return self.__playlist
def getLock(self) -> Lock:
return self.__lock
def getTextChannel(self) -> TextChannel:
return self.__textChannel
class ProcessPlayerManager(Singleton, AbstractPlayersManager):
"""
Manage all running player process, creating and storing them for future calls
Deals with the creation of shared memory
@@ -29,28 +76,51 @@ class ProcessManager(Singleton, AbstractPlayersManager):
VManager.register('Playlist', Playlist)
self.__manager = VManager()
self.__manager.start()
self.__playersProcess: Dict[int, PlayerInfo] = {}
self.__playersProcess: Dict[int, PlayerProcessInfo] = {}
self.__playersListeners: Dict[int, Tuple[Thread, bool]] = {}
self.__playersCommandsExecutor: Dict[int, ProcessCommandsExecutor] = {}
def setPlayerInfo(self, guild: Guild, info: PlayerInfo):
self.__playersProcess[guild.id] = info
def sendCommandToPlayer(self, command: VCommands, guild: Guild, forceCreation: bool = False, context: Union[Context, Interaction] = None):
if forceCreation:
processInfo = self.createPlayerForGuild(guild, context)
else:
processInfo = self.__getRunningPlayerInfo(guild)
def getOrCreatePlayerInfo(self, guild: Guild, context: Union[Context, Interaction]) -> PlayerInfo:
"""Return the process info for the guild, the user in context must be connected to a voice_channel"""
if processInfo == None:
return
queue = processInfo.getQueueToPlayer()
self.__putCommandInQueue(queue, command)
def getPlayerPlaylist(self, guild: Guild) -> Playlist:
playerInfo = self.__getRunningPlayerInfo(guild)
if playerInfo:
return playerInfo.getPlaylist()
def getPlayerLock(self, guild: Guild) -> Lock:
playerInfo = self.__getRunningPlayerInfo(guild)
if playerInfo:
return playerInfo.getLock()
def verifyIfPlayerExists(self, guild: Guild) -> bool:
return guild.id in self.__playersProcess.keys()
def createPlayerForGuild(self, guild: Guild, context: Union[Context, Interaction]) -> None:
try:
if guild.id not in self.__playersProcess.keys():
self.__playersProcess[guild.id] = self.__createProcessInfo(guild, context)
self.__playersProcess[guild.id] = self.__createProcessPlayerInfo(guild, context)
else:
# If the process has ended create a new one
if not self.__playersProcess[guild.id].getProcess().is_alive():
self.__playersProcess[guild.id] = self.__recreateProcess(guild, context)
# Start the process
self.__playersProcess[guild.id].getProcess().start()
return self.__playersProcess[guild.id]
except Exception as e:
print(f'[Error In GetPlayerContext] -> {e}')
def resetProcess(self, guild: Guild, context: Context) -> None:
def resetPlayer(self, guild: Guild, context: Context) -> None:
"""Restart a running process, already start it to return to play"""
if guild.id not in self.__playersProcess.keys():
return None
@@ -60,10 +130,10 @@ class ProcessManager(Singleton, AbstractPlayersManager):
newProcessInfo.getProcess().start() # Start the process
# Send a command to start the play again
playCommand = VCommands(VCommandsType.PLAY)
newProcessInfo.getQueueToPlayer().put(playCommand)
self.__putCommandInQueue(newProcessInfo.getQueueToPlayer(), playCommand)
self.__playersProcess[guild.id] = newProcessInfo
def getRunningPlayerInfo(self, guild: Guild) -> PlayerInfo:
def __getRunningPlayerInfo(self, guild: Guild) -> PlayerProcessInfo:
"""Return the process info for the guild, if not, return None"""
if guild.id not in self.__playersProcess.keys():
print('Process Info not found')
@@ -71,7 +141,7 @@ class ProcessManager(Singleton, AbstractPlayersManager):
return self.__playersProcess[guild.id]
def __createProcessInfo(self, guild: Guild, context: Context) -> PlayerInfo:
def __createProcessPlayerInfo(self, guild: Guild, context: Context) -> PlayerProcessInfo:
guildID: int = context.guild.id
textID: int = context.channel.id
voiceID: int = context.author.voice.channel.id
@@ -82,9 +152,9 @@ class ProcessManager(Singleton, AbstractPlayersManager):
queueToListen = Queue()
queueToSend = Queue()
process = PlayerProcess(context.guild.name, playlist, lock, queueToSend,
queueToListen, guildID, textID, voiceID, authorID)
processInfo = PlayerInfo(process, queueToSend, queueToListen,
playlist, lock, context.channel)
queueToListen, guildID, voiceID)
processInfo = PlayerProcessInfo(process, queueToSend, queueToListen,
playlist, lock, context.channel)
# Create a Thread to listen for the queue coming from the Player Process, this will redirect the Queue to a async
thread = Thread(target=self.__listenToCommands,
@@ -111,7 +181,7 @@ class ProcessManager(Singleton, AbstractPlayersManager):
except Exception as e:
print(f'[ERROR STOPPING PROCESS] -> {e}')
def __recreateProcess(self, guild: Guild, context: Union[Context, Interaction]) -> PlayerInfo:
def __recreateProcess(self, guild: Guild, context: Union[Context, Interaction]) -> PlayerProcessInfo:
"""Create a new process info using previous playlist"""
self.__stopPossiblyRunningProcess(guild)
@@ -130,8 +200,8 @@ class ProcessManager(Singleton, AbstractPlayersManager):
queueToSend = Queue()
process = PlayerProcess(context.guild.name, playlist, lock, queueToSend,
queueToListen, guildID, textID, voiceID, authorID)
processInfo = PlayerInfo(process, queueToSend, queueToListen,
playlist, lock, context.channel)
processInfo = PlayerProcessInfo(process, queueToSend, queueToListen,
playlist, lock, context.channel)
# Create a Thread to listen for the queue coming from the Player Process, this will redirect the Queue to a async
thread = Thread(target=self.__listenToCommands,
@@ -191,10 +261,18 @@ class ProcessManager(Singleton, AbstractPlayersManager):
# Set the status of this process as sleeping, only the playlist object remains
self.__playersProcess[guildID].setStatus(ProcessStatus.SLEEPING)
def __putCommandInQueue(self, queue: Queue, command: VCommands) -> None:
try:
queue.put(command)
except Exception as e:
print(f'[ERROR PUTTING COMMAND IN QUEUE] -> {e}')
async def showNowPlaying(self, guildID: int, song: Song) -> None:
commandExecutor = self.__playersCommandsExecutor[guildID]
processInfo = self.__playersProcess[guildID]
print('A')
await commandExecutor.sendNowPlaying(processInfo, song)
print('C')
class VManager(BaseManager):