TSDDCS/tsddcs-datacollect-stream/src/main/java/org/ts/ddcs/net/DataChannelUdpWithNio.java

package  org.ts.ddcs.net;

import io.netty.bootstrap.Bootstrap;
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.NioDatagramChannel;
import org.ts.ddcs.log.LogHelper;

/**
 * UDP网络连接
 * @author dylan
 */
public class DataChannelUdpWithNio {

    private String channelKey;
    private String bindAgreeement;

    private boolean nioChannelRunning = false;
    /*
    监听端口
     */
    private int port = 0;
    private EventLoopGroup bossGroup = null;

    public DataChannelUdpWithNio(String key, String agreement){
        this.channelKey = key;
        this.bindAgreeement = agreement;
    }

    public DataChannelUdpWithNio colseChannel() {
        this.nioChannelRunning = false;
        if (null != bossGroup) {
            bossGroup.shutdownGracefully();
            bossGroup = null;
        }
        LogHelper.info("******************** 关闭 NIO ********************");
        return this;
    }

    public DataChannelUdpWithNio openChannel(int port) {
        this.nioChannelRunning = true;
        this.port = port;
        LogHelper.info("******************** 启动 NIO ********************");
        return this;
    }

    public void async() {
        if (!this.nioChannelRunning) {
            LogHelper.error("调用 Async() 发生错误:未打开通道");
            return;
        }
        new DataChannelUdpWithNio.AcceptThread().start();
    }

    private class AcceptThread extends Thread {

        @Override
        public void run() {
            bossGroup = new NioEventLoopGroup();
            try {
                Bootstrap b = new Bootstrap();
                b.group(bossGroup)
                        .channel(NioDatagramChannel.class)
                        .option(ChannelOption.SO_BROADCAST, true)
                        .handler(new DatagramProverbServerHandler(channelKey,bindAgreeement));
                ChannelFuture f = b.bind(port).sync().channel().closeFuture().await();
                f.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                LogHelper.error(e);
            } finally {
                if (null != bossGroup) {
                    bossGroup.shutdownGracefully();
                }
            }
        }
    }
}