package cn.ray.gateway.session;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

/**
 * @author Ray
 * @date 2023/5/11 15:51
 * @description 数据处理器基础类
 */
public abstract class BaseHandler<T> extends SimpleChannelInboundHandler<T> {

    /**
     * SimpleChannelInboundHandler是Netty提供的一个专门用于处理传入消息的抽象类
     * 它继承自ChannelInboundHandlerAdapter
     * 并提供了一些方便的功能，如自动释放接收到的消息和处理异常等。
     * 通过继承SimpleChannelInboundHandler，我们可以重写它的channelRead0方法来处理接收到的传入消息
     * 而不必关心消息的释放和异常处理等底层细节
     * 同时，该类还提供了一些钩子方法（如channelActive和channelInactive）
     * 可以在处理消息的同时进行其他操作，如打印日志、更新状态等
     * 因此，继承SimpleChannelInboundHandler的用意是为了方便地处理传入的消息，并提高代码的可读性和可维护性
     */
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, T t) throws Exception {
        session(channelHandlerContext,channelHandlerContext.channel(),t);
    }

    protected abstract void session(ChannelHandlerContext ctx, final Channel channel, T request);
}

package cn.ray.gateway.session.handlers;

import cn.ray.gateway.session.BaseHandler;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author Ray
 * @date 2023/5/11 15:55
 * @description 会话服务处理器
 */
public class SessionServerHandler extends BaseHandler<FullHttpRequest> {

    private final Logger logger = LoggerFactory.getLogger(SessionServerHandler.class);

    @Override
    protected void session(ChannelHandlerContext ctx, Channel channel, FullHttpRequest request) {
        logger.info("网关接收请求 ===> uri: {} , method: {}",request.uri(),request.method());

        // 返回信息处理
        DefaultFullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
        // 返回信息控制
        response.content().writeBytes(JSON.toJSONBytes("访问路径被网关管理了 URI: " + request.uri(), SerializerFeature.PrettyFormat));
        // 头部信息设置
        HttpHeaders heads = response.headers();
        // 返回内容类型
        heads.add(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON + "; charset=UTF-8");
        // 响应体的长度
        heads.add(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
        // 配置长连接
        heads.add(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
        // 配置跨域访问
        heads.add(HttpHeaderNames.ACCESS_CONTROL_ALLOW_ORIGIN, "*");
        heads.add(HttpHeaderNames.ACCESS_CONTROL_ALLOW_HEADERS, "*");
        heads.add(HttpHeaderNames.ACCESS_CONTROL_ALLOW_METHODS, "GET, POST, PUT, DELETE");
        heads.add(HttpHeaderNames.ACCESS_CONTROL_ALLOW_CREDENTIALS, "true");

        channel.writeAndFlush(response);
    }
}

package cn.ray.gateway.session;

import cn.ray.gateway.session.handlers.SessionServerHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpRequestDecoder;
import io.netty.handler.codec.http.HttpResponseEncoder;

/**
 * @author Ray
 * @date 2023/5/11 16:09
 * @description 会话 channel 初始化，添加处理器
 */
public class SessionChannelInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        // HttpRequestDecoder、HttpResponseEncoder 是 Netty 本身提供的HTTP编解器
        pipeline.addLast(new HttpRequestDecoder());
        pipeline.addLast(new HttpResponseEncoder());

        // HttpObjectAggregator 用于处理除了 GET 请求外的 POST 请求时候的对象信息，否则只有上面的信息，是拿不到 POST 请求的
        pipeline.addLast(new HttpObjectAggregator(1024*1024));

        // SessionServerHandler 是我们自己实现的会话处理，用于拿到HTTP网络请求后，处理我们自己需要的业务逻辑
        pipeline.addLast(new SessionServerHandler());
    }
}

package cn.ray.gateway.session;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.concurrent.Callable;

/**
 * @author Ray
 * @date 2023/5/11 16:12
 * @description 网关会话服务
 */
public class SessionServer implements Callable<Channel> {

    private final Logger logger = LoggerFactory.getLogger(SessionServer.class);

    private final EventLoopGroup boss = new NioEventLoopGroup(1);

    private final EventLoopGroup worker = new NioEventLoopGroup();

    private Channel channel;

    @Override
    public Channel call() throws Exception {
        ChannelFuture channelFuture = null;

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(boss,worker)
                    .channel(NioServerSocketChannel.class)
                    // 设置服务器Socket的连接队列大小为128，即服务器可以同时接受128个等待处理的客户端连接请求。
                    .option(ChannelOption.SO_BACKLOG,128)
                    .childHandler(new SessionChannelInitializer());
            channelFuture = bootstrap.bind(new InetSocketAddress(7397))
                    .syncUninterruptibly();
            // 阻塞当前线程直到服务器启动完成
            // syncUninterruptibly()方法不会响应线程中断
            // 因此即使在等待操作完成的过程中，线程被中断也不会中止等待
            // 如果需要能够响应线程中断，可以使用.sync()方法，它会抛出InterruptedException异常，需要进行相应的处理。
            this.channel = channelFuture.channel();
        } catch (Exception e) {
            logger.error("socket server start error.", e);
        } finally {
            if (null!=channelFuture && channelFuture.isSuccess()) {
                logger.info("socket server start done.");
            } else {
                logger.error("socket server start error.");
            }
        }
        return channel;
    }
}

package cn.ray.gateway.test;

import cn.ray.gateway.session.SessionServer;
import io.netty.channel.Channel;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * @author Ray
 * @date 2023/5/11 16:26
 * @description
 */
public class ApiTest {

    private final Logger logger = LoggerFactory.getLogger(ApiTest.class);

    @Test
    public void test() throws ExecutionException, InterruptedException {
        SessionServer server = new SessionServer();

        Future<Channel> future = Executors.newFixedThreadPool(2).submit(server);

        Channel channel = future.get();

        if (null==channel) {
            logger.error("netty server start error, because channel is null!");
        }

        while (!channel.isActive()) {
            logger.info("NettyServer 启动服务...");
            Thread.sleep(500);
        }
        logger.info("NettyServer 启动服务成功 {}",channel.localAddress());

        Thread.sleep(Long.MAX_VALUE);
    }
}