router-backend/src/main/java/org/bdware/server/NodeCenterServer.java

package org.bdware.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.ssl.OptionalSslHandler;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.stream.ChunkedWriteHandler;
import org.apache.commons.io.FileUtils;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.core.config.Configurator;
import org.bdware.sc.DoConfig;
import org.bdware.sc.db.KeyValueDBUtil;
import org.bdware.sc.db.MultiIndexTimeRocksDBUtil;
import org.bdware.sc.db.TimeDBUtil;
import org.bdware.server.irp.LocalLHSProxy;
import org.bdware.server.nodecenter.*;
import org.bdware.server.ws.DelimiterCodec;

import java.io.File;
import java.net.MalformedURLException;
import java.net.URL;
import java.net.URLClassLoader;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

public class NodeCenterServer {
    public static final String PATH = "/NodeCenterWS";
    private static final String CONFIG_PATH = "ncconfig.json";
    private static final Logger LOGGER = LogManager.getLogger(NodeCenterServer.class);
    public static MultiIndexTimeRocksDBUtil nodeHttpLogDB =
            new MultiIndexTimeRocksDBUtil("./NodeCenterDB", NCTables.NodeHttpLog.toString());
    public static MultiIndexTimeRocksDBUtil nodeTcpLogDB =
            new MultiIndexTimeRocksDBUtil("./NodeCenterDB", NCTables.NodeTcpLog.toString());
    public static ExecutorService threadPool;
    public static ScheduledExecutorService scheduledThreadPool =
            Executors.newScheduledThreadPool(8);
    static SslContext sslContext = null;
    public static URLClassLoader pluginLoader;
//    static byte[] delimiter = "wonbifoodie".getBytes();

    static {
        KeyValueDBUtil.setupNC();
//        TimeDBUtil.setupNC();
        TimeDBUtil.setupNC();
        Configurator.setLevel(
                "io.netty.handler.codec.http.websocketx.WebSocket08FrameDecoder",
                Level.OFF);
        Configurator.setLevel(
                "io.netty.handler.codec.http.websocketx.WebSocket08FrameEncoder",
                Level.OFF);
    }

    public static void parseConf(CMDConf cmdConf) {
        LocalLHSProxy.port = cmdConf.doipPort;
        LocalLHSProxy.enabled = true;
        if (cmdConf.disableDoRepo) {
            DoConfig.callContractUsingDOI = false;
        }
        if (cmdConf.disableLocalLhs) {
            LocalLHSProxy.enabled = false;
        }
        if (!cmdConf.enableSsl.isEmpty()) {
            try {
                String[] filePaths = cmdConf.enableSsl.split(":");
                File chainedFile = new File(filePaths[0]), keyFile = new File(filePaths[1]);
                if (chainedFile.exists() && keyFile.exists()) {
                    sslContext = SslContextBuilder.forServer(chainedFile, keyFile).build();
                }
                LOGGER.debug("Enable ssl, path: " + chainedFile.getPath());
            } catch (Exception e) {
                LOGGER.warn("Enabling SSL failed.");
                e.printStackTrace();
            }
        }
        if (cmdConf.debug != null) {
            if (!cmdConf.debug.isEmpty()) {

                try {
                    String[] classes = cmdConf.debug.split(",");
                    for (String clz : classes) {
                        Configurator.setLevel(clz, Level.DEBUG);
                        LOGGER.warn("set debug: " + clz);
                    }
                } catch (Exception e) {
                    LOGGER.warn(e.getMessage());
                }
            }
        }
        if (cmdConf.overwrite) {
            cmdConf.write(CONFIG_PATH);
        }
        NodeCenterWSFrameHandler.wsPluginActions = parseStrAsList(cmdConf.wsPluginActions);
    }

    private static String[] parseStrAsList(String str) {
        if (str == null) {
            return new String[]{};
        }
        return str.split(",");
    }

    public static void main(String[] args) throws Exception {
        File confFile = new File(CONFIG_PATH);
        File confTemplate = new File(CONFIG_PATH + ".template");
        if (!confTemplate.exists()) {
            CMDConf conf = new CMDConf();
            conf.write(confTemplate.getAbsolutePath());
        }
        if (!confFile.exists() && confTemplate.exists()) {
            FileUtils.copyFile(confTemplate, confFile);
        }
        CMDConf cmf = CMDConf.parseConf(CONFIG_PATH).parseArgs(args);

        parseConf(cmf);

        if (LocalLHSProxy.enabled) {
            threadPool = Executors.newSingleThreadExecutor();
            threadPool.execute(() -> {
                try {
                    LocalLHSProxy.start();
                } catch (Exception e) {
                    LOGGER.error("local LHS proxy failed: " + e.getMessage());
                }
            });
        }

        listenSocket(cmf.servicePort + 1);
        OtherNCProxy.instance.init();

        startHttp(cmf.servicePort);
        //System.out.println("PORT"+cmf.servicePort);
    }

    public static void listenSocket(int port) throws Exception {
        LOGGER.info("listen socket at " + port);
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        ServerBootstrap b = new ServerBootstrap();
        b.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
        b.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 100)
                .localAddress(port)
                .childHandler(
                        new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel arg0) {
                                arg0.pipeline()
                                        .addLast(new DelimiterCodec())
                                        .addLast(new NodeCenterFrameHandler());
                            }
                        });
        b.bind(port).sync().channel();
    }

    public static void startHttp(int port) {
        File[] pluginJar = new File("./pluginLib/")
                .listFiles(pathname -> pathname.getName().endsWith(".jar"));
        URL[] urls;
        if (pluginJar != null && pluginJar.length > 0) {
            urls = new URL[pluginJar.length];
            for (int i = 0; i < pluginJar.length; i++) {
                try {
                    urls[i] = pluginJar[i].toURI().toURL();
                    LOGGER.info("add plugin:" + pluginJar[i].getName());
                } catch (MalformedURLException e) {
                    e.printStackTrace();
                }
            }
        } else {
            urls = new URL[]{};
        }
        pluginLoader = new URLClassLoader(urls, NodeCenterServer.class.getClassLoader());
        LOGGER.info("start at: " + port);

        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            ControlledChannelInitializer initializer = new ControlledChannelInitializer();
            b.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .localAddress(port)
                    .childHandler(initializer);
            final Channel ch = b.bind(port).sync().channel();
            ch.closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static void start(int port) {
        try {
            listenSocket(port + 1);
            NCUDPRunner.mainPort = port + 2;
            startHttp(port);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    static class ControlledChannelInitializer extends ChannelInitializer<SocketChannel> {
        NCHttpHandler handler = new NCHttpHandler();

        @Override
        protected void initChannel(SocketChannel arg0) {
            if (sslContext != null) {
                arg0.pipeline().addLast(new OptionalSslHandler(sslContext));
            } else {
                LOGGER.warn("disable ssl");
            }
            arg0.pipeline()
                    .addLast(new HttpServerCodec())
                    .addLast(new HttpObjectAggregator(65536))
                    .addLast(new WebSocketServerProtocolHandler(PATH, null, true))
                    .addLast(new ChunkedWriteHandler())
                    .addLast(handler)
                    .addLast(new NodeCenterWSFrameHandler());
        }
    }
}