server-manager/wrapper/src/index.ts

import * as amqp from "amqplib";
import os from "os";
import { loadConfig, saveConfig, RABBITMQ_URL, WRAPPER_HEARTBEAT_INTERVAL } from "./config";
import { ProcessManager } from "./processManager";
import { StatusReporter } from "./statusReporter";
import { QueryManager } from "./queryManager";
import {
  WrapperRegisterMessage,
  WrapperHeartbeatMessage,
  ServerStartCommand,
  ServerStopCommand
} from "server-manager-shared";
import { getLogger } from "./logger";

const logger = getLogger("Wrapper");


class Wrapper {
    
    processManager =
        new ProcessManager();

    queryManager =
        new QueryManager(
            this.processManager
        );

    connection!: amqp.ChannelModel;
    channel!: amqp.Channel;


    id?: string;



    async start(){


        const config =
            loadConfig();

        logger.info("Konfiguration geladen");


        if(config.wrapperId){


            this.id =
                config.wrapperId;



            logger.debug(
                "Starte mit vorhandener ID:",
                this.id
            );


        }else{


            logger.warn(
                "Keine Wrapper ID vorhanden - neue Registrierung"
            );


        }




        this.connection =
            await amqp.connect(
                RABBITMQ_URL
            );



        this.channel =
            await this.connection.createChannel();



        logger.info(
            "RabbitMQ Verbindung hergestellt"
        );



        this.queryManager.start();
        await this.register();


    }







    async register(){



        const reply =
            await this.channel.assertQueue(
                "",
                {
                    exclusive:true
                }
            );



        const replyQueue =
            reply.queue;






        await this.channel.consume(

            replyQueue,

            msg=>{


                if(!msg)
                    return;



                const data =
                    JSON.parse(
                        msg.content.toString()
                    );



                logger.debug(
                    "Antwort vom Manager:",
                    data.type
                );




                if(data.type==="wrapper.registered"){



                    this.id =
                        data.wrapperId;



                    saveConfig({

                        wrapperId:
                            this.id

                    });



                    logger.info(
                        "Registrierung beim Manager erfolgreich"
                    );
                    logger.debug(
                        "Wrapper ID gespeichert:",
                        this.id
                    );

                    if(!this.id){

                        logger.error(
                            "Keine Wrapper ID erhalten!"
                        );

                        return;

                    }


                    this.listenCommands(
                        data.commandQueue
                    );

                    const reporter =
                        new StatusReporter(

                            this.channel,

                            this.processManager,

                            this.queryManager,

                            this.id

                        );


                    reporter.start();


                }



                this.channel.ack(msg);


            }

        );








        const message: WrapperRegisterMessage = {



            type:
                "wrapper.register",



            // vorhandene ID mitsenden
            wrapperId:
                this.id,



            hostname:
                os.hostname(),



            cpu:
                os.cpus().length,



            ram:
                Math.round(
                    os.totalmem()/1024/1024
                ),



            replyQueue


        };






        await this.channel.sendToQueue(

            "wrapper.register",

            Buffer.from(
                JSON.stringify(message)
            )

        );




        logger.debug(
            "Registrierung gesendet",
            this.id ?? "ohne ID"
        );


    }







    async listenCommands(queue:string){


        await this.channel.consume(

            queue,

            msg=>{


                if(!msg)
                    return;



                const rawCommand =
                    JSON.parse(
                        msg.content.toString()
                    ) as unknown;



                const commandType = (rawCommand as { type?: string }).type;
                switch(commandType){



                    case "test": {
                        const command = rawCommand as {
                            type: string;
                            message?: unknown;
                        };

                        logger.debug(
                            "TEST COMMAND:",
                            command.message
                        );

                        break;
                    }





                    case "server.start": {
                        const command = rawCommand as ServerStartCommand;

                        this.processManager.startServer(
                            command
                        );

                        break;
                    }


                    case "server.stop": {
                        const command = rawCommand as ServerStopCommand;

                        this.processManager.stopServer(
                            command.serverId
                        );

                        break;
                    }


                    default:


                        logger.warn(
                            "Unbekannter Command:",
                            commandType
                        );


                }



                this.channel.ack(msg);



            }

        );



        logger.debug(
            "Warte auf Commands:",
            queue
        );



        this.heartbeat();

    }




    heartbeat(){

        logger.debug("Heartbeat gestartet");
        setInterval(()=>{



            if(!this.id)
                return;



            const heartbeat: WrapperHeartbeatMessage = {


                type:
                    "wrapper.heartbeat",



                wrapperId:
                    this.id


            };




            this.channel.sendToQueue(


                "wrapper.register",



                Buffer.from(

                    JSON.stringify(
                        heartbeat
                    )

                )

            );



            logger.debug(
                "Heartbeat gesendet"
            );



},WRAPPER_HEARTBEAT_INTERVAL);


    }



}



new Wrapper().start();