launcher.py

"""
This is the launcher package for server launching.
"""
from dataclasses import dataclass
from pathlib import Path
from noggin.launcher.status import LauncherStatus
import subprocess
import nice_logging as logging
from threading import Thread
from observableprops import ObservablesMeta


class ServerLauncher(metaclass=ObservablesMeta):
    observable_properties = [
        "color",
    ]

    def __init__(self,
                 jarfile: Path,
                 java_path: Path,
                 server_dir: Path,
                 server_args: list = [],
                 server_name: str = "",
                 memory: int = 1024,
                 logger: logging.logging.Logger = logging.getLogger(__name__)):
        self.jarfile = jarfile
        self.java_path = java_path
        self.server_dir = server_dir
        self.server_args = server_args
        self.server_name = server_name
        self.memory = memory
        self.logger = logger
        self.status = LauncherStatus.STOPPED

        self.logger.info(f"Initialized {self.server_name} launcher.")
        self.check_server_dir()

    def launch(self):
        self.logger.info(f"Launching {self.server_name}...")

        cmd = [str(self.java_path), "-Xmx{}M".format(self.memory), "-jar", str(self.jarfile)] + self.server_args
        self.logger.debug("Generated command: " + " ".join(cmd))
        self.logger.info("Running launch command...")

        self.process = subprocess.Popen(cmd, cwd=self.server_dir, stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE)
        self.status = LauncherStatus.RUNNING

        self.logger.info(f"{self.server_name} launched with PID {self.process.pid}")

        self.watcher_thread = Thread(target=self._watch_process)
        self.watcher_thread.start()

    def stop(self):
        if self.status != LauncherStatus.RUNNING:
            self.logger.warning(f"Cannot stop {self.server_name} as it is not running.")
            return
        self.logger.info(f"Stopping {self.server_name}...")
        self.process.terminate()
        self.status = LauncherStatus.STOPPED
        self.logger.info(f"{self.server_name} stopped successfully.")
    
    def check_server_dir(self):
        """
        Check if the server directory exists. If not, create it.
        """
        if not self.server_dir.exists():
            self.logger.info(f"Server directory {self.server_dir} does not exist. Creating the folder.")
            self.server_dir.mkdir(parents=True)
        else:
            self.logger.info(f"Server directory {self.server_dir} exists.")

    def _watch_process(self):
        """
        Watch the process to update the server status.
        """
        self.logger.debug("Watcher thread started.")
        self.process.wait()

        if self.process.returncode == 0:
            self.status = LauncherStatus.STOPPED
            self.logger.info(f"{self.server_name} has stopped with code 0.")
        else:
            self.status = LauncherStatus.CRASHED
            self.logger.warning(f"{self.server_name} has crashed with code {self.process.returncode}")


"""
>>> launcher.stdin.write(b"stop\n")
>>> launcher.stdin.flush()
"""