rosetta-wss/src/main/java/io/orprotocol/Server.java

package io.orprotocol;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.java_websocket.WebSocket;
import org.java_websocket.handshake.ClientHandshake;
import org.java_websocket.server.WebSocketServer;

import io.orprotocol.client.Client;
import io.orprotocol.packet.Packet;
import io.orprotocol.packet.PacketExecutor;
import io.orprotocol.packet.PacketManager;

public class Server extends WebSocketServer {

    private PacketManager packetManager;
    private Settings settings;
    private ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
    private Context context;
    private ServerListener listener;

    /**
     * Конструктор сервера
     * @param settings базовые настройки серверера
     * @param packetManager менеджер пакетов (обработчиков и зарегистрированных пакетов)
     */
    public Server(Settings settings, PacketManager packetManager) {
        super(new InetSocketAddress(settings.port));
        this.settings = settings;
        this.packetManager = packetManager;
    }

    /**
     * Конструктор сервера с объектом прикрепления
     * @param settings базовые настройки серверера
     * @param packetManager менеджер пакетов (обработчиков и зарегистрированных пакетов)
     * @param context вложение которое будет передаваться всем серрверным обработчикам пакетов, 
     * может быть использовано для передачи контекста приложения
     */
    public Server(Settings settings, PacketManager packetManager, Context context) {
        super(new InetSocketAddress(settings.port));
        this.settings = settings;
        this.packetManager = packetManager;
        this.context = context;
    }

    public Server(Settings settings, PacketManager packetManager, Context context, ServerListener listener) {
        super(new InetSocketAddress(settings.port));
        this.settings = settings;
        this.packetManager = packetManager;
        this.context = context;
        this.listener = listener;
    }

    @Override
    public void onClose(WebSocket socket, int arg1, String arg2, boolean arg3) {
        if(this.listener == null){
            return;
        }
        this.listener.onClientDisconnect(this, socket.getAttachment());
    }

    @Override
    public void onError(WebSocket arg0, Exception arg1) {
        if(this.listener == null){
            return;
        }
        this.listener.onError(this, arg1);
    }

    @Override
    public void onMessage(WebSocket socket, String message) {
        /**
         * Обновляем время последнего полученного heartbeat.
         * Так как клиент отпраивл нам пакет, он живой.
         */
        Client client = socket.getAttachment();
        client.updateHeartbeat();
        if(this.listener != null) {
            /**
             * Сообщение от клиента, но пакет не сформирован, передаем null
             */
            this.listener.onPacketReceived(this, client, null);
        }
    }

    @Override
    public void onMessage(WebSocket socket, ByteBuffer byteBuffer) {
        Client client = socket.getAttachment();
        byte[] bytes = byteBuffer.array();
        Stream stream = new Stream(bytes);
        int packetId = stream.readInt16();
        /**
         * Обновляем время последнего полученного heartbeat.
         * Так как клиент отпраивл нам пакет, он живой.
         */
        client.updateHeartbeat();

        if(!this.packetManager.hasPacketSupported(packetId)){
            /**
             * Если пакет не поддерживается, отключаем клиента с соответствующим кодом ошибки.
             */
            client.disconnect(ServerFailures.UNSUPPORTED_PACKET);
            return;
        }
        if(!this.packetManager.hasExecutorDelegated(packetId)){
            /**
             * Если для пакета не назначен обработчик, отключаем клиента с соответствующим кодом ошибки.
             */
            client.disconnect(ServerFailures.UNSUPPORTED_PACKET);
            return;
        }
        Class<? extends Packet> packetClass = this.packetManager.getPacketClass(packetId);

        try {
            Packet packet = packetClass.getConstructor().newInstance();
            packet.packetId = packetId;
            /**
             * Читаем данные пакета из потока.
             */
            packet.read(stream);
            /**
             * Получаем обработчик пакета и вызываем его метод обработки.
             */
            Class<? extends PacketExecutor> executorClass = this.packetManager.getExecutors().get(packetId);
            PacketExecutor executor = executorClass.getConstructor().newInstance();
            executor.settings = this.settings;
            executor.context = this.context;
            if(listener != null && listener.onPacketReceived(this, client, packet)) {
                /**
                 * Если слушатель сервера вернул true, пакет не обрабатываем.
                 */
                return;
            }
            executor.onPacketReceived(packet, client);
        } catch (Exception e) {
            System.out.println("Error while processing packet " + packetClass.getName());
            System.out.println(e.getStackTrace());
        }
    }    

    @Override
    public void onOpen(WebSocket socket, ClientHandshake arg1) {
        /**
         * Создаем нового клиента при открытии соединения.
         * Передаем интервал heartbeat из настроек сервера.
         * Если клиент не отправляет heartbeat в указанный интервал, его можно отключить.
         */
        Client client = new Client(socket, this.settings.heartbeatInterval);
        socket.setAttachment(client);
        if(this.listener == null){
            return;
        }
        if(this.listener.onClientConnect(this, client)) {
            client.disconnect(ServerFailures.SERVER_NOT_ACCEPT_CLIENT);
            return;
        }
    }

    @Override
    public void onStart() {
        /**
         * Настраиваем планировщик для проверки активности клиентов.
         */
        this.inactivityShedulerTweaker();

        int port = this.getPort();
        System.out.println("\u001B[32mServer started at x.x.x.x:" + port + "\u001B[0m");
        if(this.listener == null){
            return;
        }
        this.listener.onServerStart(this);
    }

    /**
     * Планировщик для проверки активности клиентов.
     * Если планировщик обнаруживает неактивного клиента, он отключает его с соответствующим кодом ошибки.
     */
    public void inactivityShedulerTweaker() {
        this.scheduler.scheduleAtFixedRate(() -> {
            for(WebSocket socket : this.getConnections()) {
                Client client = socket.getAttachment();
                if(!client.isAlive()) {
                    client.disconnect(ServerFailures.INACTIVITY_TIMEOUT);
                }
            }
        }, this.settings.heartbeatInterval, this.settings.heartbeatInterval, TimeUnit.MILLISECONDS);
    }

}