Java 类io.netty.channel.Channel 实例源码

public void scanNotActiveChannel(){
    try {
         if (this.groupChannelLock.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
             for(Entry<String,Set<Channel>> entry : groupChannelTable.entrySet()){
                 if(CollectionUtils.isEmpty(entry.getValue())){
                     continue;
                 }
                 Iterator<Channel> it = entry.getValue().iterator();
                 while(it.hasNext()){
                     Channel c = it.next();
                     if(!c.isActive()){
                         it.remove();
                     }
                 }
             }
         }else {
             log.warn("ProducerManager scanNotActiveChannel lock timeout");
         }
} catch (Exception e) {
    log.error("scanNotActiveChannel",e);
}finally{
    this.groupChannelLock.unlock();
}
  }

public NetworkDispatcher connectToLocal(SocketAddress address)
{
    NetworkDispatcher dispatch = new NetworkDispatcher(this, NetworkSide.CLIENT);

    final EventLoopGroup boss = new DefaultEventLoopGroup();
    final Bootstrap b = new Bootstrap()
            .group(boss)
            .handler(new ChannelInitializer<Channel>() {
                @Override
                protected void initChannel(Channel ch) throws Exception
                {
                    ch.pipeline().addLast(dispatch);
                }
            })
            .channel(LocalChannel.class);

    //Connect and wait until done
    b.connect(address).syncUninterruptibly();

    return dispatch;
}

public static boolean start(Channel channel, String name, Logger logger) {
    final PacketEncoder oldEncoder = channel.pipeline().get(PacketEncoder.class);
    final PacketDecoder oldDecoder = channel.pipeline().get(PacketDecoder.class);

    channel.eventLoop().execute(() -> {
        if(channel.isOpen()) {
            if(oldEncoder != null) {
                channel.pipeline().replace(oldEncoder, "encoder", new Encoder(logger, name));
            }

            if(oldDecoder != null) {
                channel.pipeline().replace(oldDecoder, "decoder", new Decoder(logger, name));
            }
        }
    });

    return oldEncoder != null || oldDecoder != null;
}

public ReplyMsg writeAndSync(final Channel channel, final AskMsg askMsg, final long timeout) throws Exception {

        if (channel == null) {
            throw new NullPointerException("channel");
        }
        if (askMsg == null) {
            throw new NullPointerException("askMsg");
        }
        if (timeout <= 0) {
            throw new IllegalArgumentException("timeout <= 0");
        }

        String requestId = UUID.randomUUID().toString();
        askMsg.setRequestId(requestId);

        WriteFuture<BaseMsg> future = new SyncWriteFuture(askMsg.getRequestId());
        SyncWriteMap.syncKey.put(askMsg.getRequestId(), future);

        System.out.println("发起请求，请求id：" + requestId + "，请求参数：" + askMsg.getData());

        ReplyMsg response = doWriteAndSync(channel, askMsg, timeout, future);

        SyncWriteMap.syncKey.remove(askMsg.getRequestId());
        return response;
    }

@Override
protected void handleRegisterRequest(RegisterRequestMessage request,
                                     Channel channel) {
    try {
        Scope scope = TProtocolUtil.getScope(request.store.getScope());
        if (request.store.isPersist())
            syncManager.registerPersistentStore(request.store.storeName,
                                                scope);
        else
            syncManager.registerStore(request.store.storeName, scope);
        RegisterResponseMessage m = new RegisterResponseMessage();
        AsyncMessageHeader header = new AsyncMessageHeader();
        header.setTransactionId(request.getHeader().getTransactionId());
        m.setHeader(header);
        SyncMessage bsm =
                new SyncMessage(MessageType.REGISTER_RESPONSE);
        bsm.setRegisterResponse(m);
        channel.writeAndFlush(bsm);
    } catch (Exception e) {
        channel.writeAndFlush(getError(request.getHeader().getTransactionId(), e,
                               MessageType.REGISTER_REQUEST));
    }
}

@Override
public void preCreateConnections(final int count)
throws ConnectException, IllegalArgumentException {
    if(count > 0) {
        for(int i = 0; i < count; i ++) {
            final Channel conn = connectToAnyNode();
            if(conn == null) {
                throw new ConnectException(
                    "Failed to pre-create the connections to the target nodes"
                );
            }
            final String nodeAddr = conn.attr(ATTR_KEY_NODE).get();
            if(conn.isActive()) {
                final Queue<Channel> connQueue = availableConns.get(nodeAddr);
                if(connQueue != null) {
                    connQueue.add(conn);
                }
            } else {
                conn.close();
            }
        }
        LOG.info("Pre-created " + count + " connections");
    } else {
        throw new IllegalArgumentException("Connection count should be > 0, but got " + count);
    }
}

protected void handshake(HelloMessage request, Channel channel) {
    try {
        switch (getAuthScheme()) {
            case CHALLENGE_RESPONSE:
                handshakeChallengeResponse(request, channel);
                break;
            case NO_AUTH:
                // shouldn't get here
                break;
        }
    } catch (AuthException e) {
        logger.warn("[{}->{}] Failed to authenticate connection: {}",
                    new Object[]{getLocalNodeIdString(), 
                                 getRemoteNodeIdString(), 
                                 e.getMessage()});
        channel.writeAndFlush(getError(request.getHeader().getTransactionId(), 
                               e, MessageType.HELLO));
        channel.close();
    }
}

/**
 * Adds a channel that listens locally
 */
public SocketAddress addLocalEndpoint()
{
    ChannelFuture channelfuture;

    synchronized (this.endpoints)
    {
        channelfuture = ((ServerBootstrap)((ServerBootstrap)(new ServerBootstrap()).channel(LocalServerChannel.class)).childHandler(new ChannelInitializer<Channel>()
        {
            protected void initChannel(Channel p_initChannel_1_) throws Exception
            {
                NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.SERVERBOUND);
                networkmanager.setNetHandler(new NetHandlerHandshakeMemory(NetworkSystem.this.mcServer, networkmanager));
                NetworkSystem.this.networkManagers.add(networkmanager);
                p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
            }
        }).group((EventLoopGroup)eventLoops.getValue()).localAddress(LocalAddress.ANY)).bind().syncUninterruptibly();
        this.endpoints.add(channelfuture);
    }

    return channelfuture.channel().localAddress();
}

/**
 * 对准备接入的 Channel 做进一步处理
 *
 * @param channel 准备接入的 Channel
 */
public void accessibleChannel(Channel channel) {
    String id = channel.id().asLongText();
    logger.info("「Channel」" + "新的 Channel 接入 [" + id + "]");
    CHANNEL_MAP.put(id, -1);
    HASHED_WHEEL_TIMER.newTimeout(task -> {
        Integer index = CHANNEL_MAP.get(id);
        if (index == -1) {
            logger.warn("「Channel」" + "新的 Channel 未反馈 ID [" + id + "]");
            channel.disconnect();
        } else if (index > 0 && index <= DeviceSetting.MAX_GROUP_ID) {
            SENDING_MESSAGE_QUEUE.get(index).clear();
            Channel oldChannel = CHANNEL_ARRAY.get(index);
            if (oldChannel != null && oldChannel.isActive()) {
                manualRemoveChannel(CHANNEL_ARRAY.get(index));
                manualRemoveChannel(channel);
                logger.warn("「Channel」" + "新的 Channel 欲覆盖已激活的 Channel [" + id + "]");
            } else {
                CHANNEL_ARRAY.set(index, channel);
                logger.info("「Channel」" + "新的 Channel「" + index + "」已成功装配 [" + id + "]");
            }
        } else {
            logger.warn("「Channel」" + "新的 Channel 装配出错 [" + id + "]");
        }
    }, CommSetting.ACCESSIBLE_CHANNEL_REPLY_INTERVAL, TimeUnit.SECONDS);
}

private static Pair<EventLoopGroup, Class<? extends Channel>> createEventLoopGroup(
    Configuration conf) {
  // Max amount of threads to use. 0 lets Netty decide based on amount of cores
  int maxThreads = conf.getInt(CLIENT_MAX_THREADS, 0);

  // Config to enable native transport. Does not seem to be stable at time of implementation
  // although it is not extensively tested.
  boolean epollEnabled = conf.getBoolean(USE_NATIVE_TRANSPORT, false);

  // Use the faster native epoll transport mechanism on linux if enabled
  if (epollEnabled && JVM.isLinux() && JVM.isAmd64()) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Create EpollEventLoopGroup with maxThreads = " + maxThreads);
    }
    return new Pair<EventLoopGroup, Class<? extends Channel>>(new EpollEventLoopGroup(maxThreads,
        Threads.newDaemonThreadFactory("AsyncRpcChannel")), EpollSocketChannel.class);
  } else {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Create NioEventLoopGroup with maxThreads = " + maxThreads);
    }
    return new Pair<EventLoopGroup, Class<? extends Channel>>(new NioEventLoopGroup(maxThreads,
        Threads.newDaemonThreadFactory("AsyncRpcChannel")), NioSocketChannel.class);
  }
}

/**
     * 适配
     */
    @Override
    protected ChannelHandler fixHandlerBeforeConnect(final ChannelHandler handler) {
        ChannelHandler result=new ShareableChannelInboundHandler() {
            @Override
            public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
                Channel ch=ctx.channel();
                ch.pipeline().addLast(new HttpClientCodec());
                ch.pipeline().addLast(new HttpObjectAggregator(64*1024));
                ch.pipeline().addLast(new WebSocketClientProtocolHandler(WebSocketClientHandshakerFactory.newHandshaker(uri, WebSocketVersion.V13, null, false, new DefaultHttpHeaders())));
                ch.pipeline().addLast(new WebSocketConnectedClientHandler(handler));
                ctx.pipeline().remove(this);//移除当前handler
                ctx.pipeline().fireChannelRegistered();//重新从第一个handler抛出事件
            }
        };
//      ChannelInitializer<SocketChannel> result=new ChannelInitializer<SocketChannel>() {
//            @Override
//            protected void initChannel(SocketChannel ch) {
//              ch.pipeline().addLast(new HttpClientCodec());
//              ch.pipeline().addLast(new HttpObjectAggregator(64*1024));
//              ch.pipeline().addLast(new WebSocketClientProtocolHandler(WebSocketClientHandshakerFactory.newHandshaker(uri, WebSocketVersion.V13, null, false, new DefaultHttpHeaders())));
//              ch.pipeline().addLast(new WebSocketConnectedClientHandler(handler));
//            }
//        };
        return result;
    }

/**
 * Broker主动通知Consumer，Id列表发生变化，Oneway
 */
public void notifyConsumerIdsChanged(
    final Channel channel,
    final String consumerGroup) {
    if (null == consumerGroup) {
        log.error("notifyConsumerIdsChanged consumerGroup is null");
        return;
    }

    NotifyConsumerIdsChangedRequestHeader requestHeader = new NotifyConsumerIdsChangedRequestHeader();
    requestHeader.setConsumerGroup(consumerGroup);
    RemotingCommand request =
        RemotingCommand.createRequestCommand(RequestCode.NOTIFY_CONSUMER_IDS_CHANGED, requestHeader);

    try {
        this.brokerController.getRemotingServer().invokeOneway(channel, request, 10);
    } catch (Exception e) {
        log.error("notifyConsumerIdsChanged exception, " + consumerGroup, e.getMessage());
    }
}

public List<String> getAllClientId() {
    List<String> result = new ArrayList<>();

    Iterator<Entry<Channel, ClientChannelInfo>> it = this.channelInfoTable.entrySet().iterator();

    while (it.hasNext()) {
        Entry<Channel, ClientChannelInfo> entry = it.next();
        ClientChannelInfo clientChannelInfo = entry.getValue();
        result.add(clientChannelInfo.getClientId());
    }

    return result;
}

/**
 * 异步调用
 * @param channel
 * @param request
 * @param timeoutMillis
 * @param invokeCallback
 * @throws InterruptedException
 * @throws RemotingTooMuchRequestException
 * @throws RemotingTimeoutException
 * @throws RemotingSendRequestException
 */
@SuppressWarnings("rawtypes")
public void invokeAsyncImpl(final Channel channel, final RemotingProtocol request, final long timeoutMillis, final InvokeCallback invokeCallback) throws InterruptedException, RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException {
    final long opaque = request.getOpaque();
    //        boolean acquired = this.semaphoreAsync.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
    final SemaphoreOnce once = new SemaphoreOnce(this.semaphoreAsync);
    final ResponseFuture responseFuture = new ResponseFuture(opaque, timeoutMillis, invokeCallback, once);
    responseTable.put(opaque, responseFuture);
    try {
        channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture f) throws Exception {
                //此步代表发送操作成功   设置的sendrequest值是为了区分发送失败还是服务端处理失败的
                if (f.isSuccess()) {
                    responseFuture.setSendRequestOK(true);
                    return;
                } else {
                    responseFuture.setSendRequestOK(false);
                }
                responseFuture.putResponse(null);
                responseTable.remove(opaque);
                //                    try {
                //                        executeInvokeCallback(responseFuture);
                //                    } catch (Throwable e) {
                //                        logger.warn("excute callback in writeAndFlush addListener, and callback throw", e);
                //                    } finally {
                //                        responseFuture.release();
                //                    }
                logger.warn("send a request command to channel <{}> failed.", RemotingHelper.parseChannelRemoteAddr(channel));
            }
        });
    } catch (Exception e) {
        responseFuture.release();
        logger.warn("send a request command to channel <" + RemotingHelper.parseChannelRemoteAddr(channel) + "> Exception", e);
        throw new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), e);
    }

}

public static void registerClienId(String clientId, Channel chn) {
  if (chn == null) {
    return;
  }
  if (clientId == null) {
    return;
  }
  chn.closeFuture().addListener(clientRemover);
  channelClientIdMap.put(chn, clientId);
  ChannelEntity oldChannel = cientIdChannelMap.put(clientId, new TcpChannelEntity(chn));
  if (oldChannel != null) {
    removeChannel(oldChannel.getChannel());
    oldChannel.getChannel().close();
  }
}

private RemotingCommand getProducerConnectionList(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    final RemotingCommand response = RemotingCommand.createResponseCommand(null);
    final GetProducerConnectionListRequestHeader requestHeader =
            (GetProducerConnectionListRequestHeader) request.decodeCommandCustomHeader(GetProducerConnectionListRequestHeader.class);

    ProducerConnection bodydata = new ProducerConnection();
    HashMap<Channel, ClientChannelInfo> channelInfoHashMap =
            this.brokerController.getProducerManager().getGroupChannelTable().get(requestHeader.getProducerGroup());
    if (channelInfoHashMap != null) {
        Iterator<Map.Entry<Channel, ClientChannelInfo>> it = channelInfoHashMap.entrySet().iterator();
        while (it.hasNext()) {
            ClientChannelInfo info = it.next().getValue();
            Connection connection = new Connection();
            connection.setClientId(info.getClientId());
            connection.setLanguage(info.getLanguage());
            connection.setVersion(info.getVersion());
            connection.setClientAddr(RemotingHelper.parseChannelRemoteAddr(info.getChannel()));

            bodydata.getConnectionSet().add(connection);
        }

        byte[] body = bodydata.encode();
        response.setBody(body);
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }

    response.setCode(ResponseCode.SYSTEM_ERROR);
    response.setRemark("the producer group[" + requestHeader.getProducerGroup() + "] not exist");
    return response;
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, Packet packet) throws Exception {
    Channel channel = ctx.channel();
    Server server = Cloud.getInstance().getServerManager().getChannelToServer().get(channel);

    if (packet instanceof PacketPing && server != null) {
        PacketPing pingPacket = (PacketPing) packet;
        long ping = System.currentTimeMillis() - pingPacket.getTimestamp();
        server.setPing(ping);
    } else if (packet instanceof PacketServerRegister) {
        PacketServerRegister registerPacket = (PacketServerRegister) packet;
        switch (registerPacket.getType()) {
            case BUNGEECORD:
                BungeeServer bungeeServer = new BungeeServer(channel);
                bungeeServer.setPrefix("bungee");
                Cloud.getInstance().getServerManager().registerServer(bungeeServer);
                break;
            case SPIGOT:
                SpigotServer spigotServer = new SpigotServer(channel);
                spigotServer.setPrefix(registerPacket.getPrefix());
                spigotServer.setBukkitPort(registerPacket.getBukkitPort());
                Cloud.getInstance().getServerManager().registerServer(spigotServer);
                break;
            case DAEMON:
                Daemon daemon = new Daemon(channel);
                daemon.setPrefix("daemon");
                Cloud.getInstance().getServerManager().registerServer(daemon);
                break;
            default:
        }
    } else if (packet instanceof PacketCloseConnection) {
        channel.close();
    } else if (packet instanceof PacketServerLoad) {
        PacketServerLoad serverLoad = (PacketServerLoad) packet;
        Cloud.getInstance().getEventManager().callEvent(new DaemonLoadEvent(serverLoad.getCpuLoad(), serverLoad.getFreeRam(), server));
    }

    Cloud.getInstance().getEventManager().callEvent(new PacketReceivingEvent(packet, server));

}

static void addSession(String sessionId, Channel channel) {
    Map<String, HttpSession> map = sessions();
    if (map.get(sessionId) == null) {
        HttpSession httpSession = new HttpSession();
        httpSession.setSessionId(sessionId);
        httpSession.setChannel(channel);
        map.put(sessionId, httpSession);
    }
}

@Override
public void connectionTerminated(Channel channel) {
  // Close all streams which have been associated with the channel.
  for (Map.Entry<Long, StreamState> entry: streams.entrySet()) {
    StreamState state = entry.getValue();
    if (state.associatedChannel == channel) {
      streams.remove(entry.getKey());

      // Release all remaining buffers.
      while (state.buffers.hasNext()) {
        state.buffers.next().release();
      }
    }
  }
}

@Override
public void onEvent(Object[] elements) throws Exception {
    BaseMessage msg;
    Map<Channel,List<BaseMessage>> msgMap = new ConcurrentHashMap<Channel, List<BaseMessage>>();
    for(Object obj:elements){
        msg = (BaseMessage) obj;
        List<BaseMessage> msgList = msgMap.get(msg.getChannel());
        if(msgList == null){
            msgList = new LinkedList<BaseMessage>();
            msgMap.put(msg.getChannel(),msgList);
        }
        msgList.add(msg);

    }

    // Traverse the map
    for (Channel keyChannel : msgMap.keySet()) {
        if (keyChannel == null) {
            logger.error("Channel {} have been destoryed/removed for case of connection been close!", keyChannel);
            return;
        }
        List<BaseMessage> msgList1 = msgMap.get(keyChannel);
        if (logger.isTraceEnabled()) {
            logger.trace("get channel here::{}", keyChannel);
        }
        for (BaseMessage msgIns : msgList1) {
            keyChannel.write(msgIns, keyChannel.voidPromise());
        }
        keyChannel.flush();
    }

}

public void destroy() {
  logger.info("MQTT server is stopping...");
  for (Channel channel : channels) {
    channel.close();
  }
  bossGroup.shutdownGracefully();
  workerGroup.shutdownGracefully();
  logger.info("MQTT server is stopped...");
}

private Channel getAndCreateChannel(final String addr) throws InterruptedException {
    if (null == addr)
        return getAndCreateNameserverChannel();

    ChannelWrapper cw = this.channelTables.get(addr);
    if (cw != null && cw.isOK()) {
        return cw.getChannel();
    }

    return this.createChannel(addr);
}

@Override
public boolean saveClientInfo(Protocol protocol, Channel channel) {
    if (!(protocol instanceof CommonProtocol)) {
        logger.error("[error] >>> protocol is not correctly");
        return false;
    }


    String clientJson = new String(((CommonProtocol) protocol).getBody());
    ClientInfo info = JSON.parseObject(clientJson, ClientInfo.class);

    logger.info("[info] >>> current sync client info: \n{}", JSON.toJSONString(info, true));

    boolean isCache = false;

    if (!(isCache = DNCSContext.CLIENT_INFO_MAP.containsKey(info.getNodeName()))) {
        synchronized (DNCSContext.class) {
            for (int i = 0; i < info.getNodeCore(); i++) {
                CoreChannel coreChannel = new CoreChannel();
                coreChannel.setChannel(channel);
                coreChannel.setNodeName(info.getNodeName());
                DNCSContext.CORE_LIST.add(coreChannel);
            }
        }

        logger.info("[info] >>> client total count: {}", DNCSContext.CLIENT_TOTAL.incrementAndGet());
        logger.info("[info] >>> core total count: {}", DNCSContext.CORE_TOTAL.addAndGet(info.getNodeCore()));
    }

    DNCSContext.CLIENT_INFO_MAP.put(info.getNodeName(), info);
    DNCSContext.CLIENT_CHANNEL_MAP.put(info.getNodeName(), channel);

    if (isCache) {
        logger.info("[info] >>> client total count: {}", DNCSContext.CLIENT_TOTAL.get());
        logger.info("[info] >>> core total count: {}", DNCSContext.CORE_TOTAL.get());
    }

    return true;
}

@Test
public void channelRead0() throws Exception {
    Channel channel = Mockito.mock(Channel.class);
    PeerExplorer peerExplorer = Mockito.mock(PeerExplorer.class);
    UDPChannel udpChannel = new UDPChannel(channel, peerExplorer);

    DiscoveryEvent event = Mockito.mock(DiscoveryEvent.class);
    udpChannel.channelRead0(Mockito.mock(ChannelHandlerContext.class), event);

    Mockito.verify(peerExplorer, Mockito.times(1)).handleMessage(event);
}

public static void closeChannel(Channel channel) {
    final String addrRemote = RemotingHelper.parseChannelRemoteAddr(channel);
    channel.close().addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            logger.info("closeChannel: close the connection to remote address[{}] result: {}", addrRemote, future.isSuccess());
        }
    });
}

@SuppressWarnings("unchecked")
private void registerChannelHandler() {
    Object mcServer = getMinecraftServer.get(Bukkit.getServer());
    Object serverConnection = getServerConnection.get(mcServer);
    boolean looking = true;

    // We need to synchronize against this list
    networkManagers = (List<Object>) getNetworkMarkers.invoke(null, serverConnection);
    createServerChannelHandler();

    // Find the correct list, or implicitly throw an exception
    for (int i = 0; looking; i++) {
        List<Object> list = TReflection.getField(serverConnection.getClass(), List.class, i).get(serverConnection);

        for (Object item : list) {
            if (!ChannelFuture.class.isInstance(item))
                break;

            // Channel future that contains the server connection
            Channel serverChannel = ((ChannelFuture) item).channel();

            serverChannels.add(serverChannel);
            serverChannel.pipeline().addFirst(serverChannelHandler);
            looking = false;
        }
    }
}

public void channelAcquired(Channel channel) throws Exception {
    if (LOG.isDebugEnabled()) {
        LOG.debug("channel acquired : {}", channel.toString());
    }

    channel.pipeline().get(FastdfsHandler.class).operation(null);
}

public NetworkDispatcher connectToPublic(InetAddress address, int port)
{
    NetworkDispatcher dispatch = new NetworkDispatcher(this, NetworkSide.CLIENT);

    final EventLoopGroup boss = new NioEventLoopGroup();
    final Bootstrap b = new Bootstrap()
            .group(boss)
            .handler(new ChannelInitializer<Channel>() {
                @Override
                protected void initChannel(Channel ch) throws Exception
                {
                    final PacketRegistry registry = NetworkEngine.this.packetRegistry;

                    ch.pipeline()
                            .addLast(new VarInt21FrameDecoder())
                            .addLast(new PacketDecoder(NetworkSide.CLIENT,
                                    registry))
                            .addLast(new VarInt21FrameEncoder())
                            .addLast(new PacketEncoder(NetworkSide.SERVER,
                                    registry))
                            .addLast(dispatch);
                }
            })
            .channel(NioSocketChannel.class);

    //Connect and wait until done
    b.connect(address, port).syncUninterruptibly();

    return dispatch;
}

/**
 * 添加Channel
 * @param channel
 */
public void addChannel(Channel channel){
    String remoteAddr = NettyUtil.parseChannelRemoteAddr(channel);
    if (!channel.isActive())  logger.error("channel is not active, address: {}", remoteAddr);
    ChatUser chatUser = new ChatUser();
    chatUser.setAddr(remoteAddr);
    chatUser.setChannel(channel);
    chatUserMap.put(channel,chatUser);
}

public NioConnDroppingServer(final int port, final int dropEveryRequest)
throws InterruptedException {
    dispatchGroup = new NioEventLoopGroup();
    workerGroup = new NioEventLoopGroup();
    final ServerBootstrap bootstrap = new ServerBootstrap()
        .group(dispatchGroup, workerGroup)
        .channel(NioServerSocketChannel.class)
        .childHandler(
            new ChannelInitializer<SocketChannel>() {
                @Override
                public final void initChannel(final SocketChannel ch) {
                    ch.pipeline().addLast(
                        new SimpleChannelInboundHandler<Object>() {
                            @Override
                            protected final void channelRead0(
                                final ChannelHandlerContext ctx, final Object msg
                            ) throws Exception {
                                if(0 == reqCounter.incrementAndGet() % dropEveryRequest) {
                                    final Channel conn = ctx.channel();
                                    System.out.println("Dropping the connection " + conn);
                                    conn.close();
                                }
                            }
                        }
                    );
                }
            }
        );

    bindFuture = bootstrap.bind(port).sync();
}

@Override
public String telnet(Channel channel, String message) {
    if (message == null || message.length() == 0) {
        return getDescription();
    }
    if (message.equalsIgnoreCase("scheduled")) {
        ScheduledService.reset();
        return "Reset scheduled operation has been send to server.";
    } else {
        return getDescription();
    }
}

public static String parseChannelRemoteName(final Channel channel) {
    if (null == channel) {
        return "";
    }
    final InetSocketAddress remote = (InetSocketAddress) channel.remoteAddress();
    if (remote != null) {
        return remote.getAddress().getHostName();
    }
    return "";
}

private void clearUnheart() {
    if (checkHeartWhenConnectionCount > connectionCount.get()) {
        return;
    }
    long now = System.currentTimeMillis();
    for (Channel channel : validatedChannels) {
        long last = 0;
        long first = 0;
        int count = 0;
        if (channel.hasAttr(LAST_HEART_TIME)) {
            last = channel.attr(LAST_HEART_TIME).get();
        }
        if (channel.hasAttr(FIRST_HEART_TIME)) {
            first = channel.attr(FIRST_HEART_TIME).get();
        }
        if (channel.hasAttr(HEART_COUNT)) {
            count = channel.attr(HEART_COUNT).get();
        }
        int allow = (int) ((now - first) / (heartIntervalSec * 1000));
        if (count - 2 > allow) {
            log.error(channel + " heart too quick,might be Game Accelerator,please check!");
            channel.pipeline().fireExceptionCaught(new HeartTooQuickException(channel, first, now, count, allow));
            channel.attr(FIRST_HEART_TIME).set(now);
            channel.attr(HEART_COUNT).set(0);
        }
        if (count < allow - 2) {
            channel.pipeline().fireExceptionCaught(new HeartNotAnswerException(channel, first, last, count));
            channel.close();
        }
    }
}

public static void closeChannel(Channel channel) {
    final String addrRemote = RemotingHelper.parseChannelRemoteAddr(channel);
    channel.close().addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            log.info("closeChannel: close the connection to remote address[{}] result: {}", addrRemote,
                future.isSuccess());
        }
    });
}

@Override
public void run()
{
    System.out.println("Starting Server...");
    EventLoopGroup boss = new NioEventLoopGroup();
    EventLoopGroup worker = new NioEventLoopGroup();
    try
    {
        ServerBootstrap b = new ServerBootstrap();
        b.group(boss, worker)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ServerChannelInitializer());

        System.out.println("Connecting...");
        Channel ch = b.bind(ChubbyCat.PORT).sync().channel();
        System.out.println("Connection Established!");

        //Process stuff

        ch.closeFuture().sync();
    }
    catch (Exception e)
    {
        e.printStackTrace();
    }
    finally
    {
        System.out.println("Stopping Server...");
        boss.shutdownGracefully();
        worker.shutdownGracefully();
    }

    System.out.println("Connection Ended!");
}

public static void closeChannel(Channel channel) {
    final String addrRemote = RemotingHelper.parseChannelRemoteAddr(channel);
    channel.close().addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            log.info("closeChannel: close the connection to remote address[{}] result: {}", addrRemote,
                    future.isSuccess());
        }
    });
}

public ClientChannelInfo findChannel(final String clientId) {
    Iterator<Entry<Channel, ClientChannelInfo>> it = this.channelInfoTable.entrySet().iterator();
    while (it.hasNext()) {
        Entry<Channel, ClientChannelInfo> next = it.next();
        if (next.getValue().getClientId().equals(clientId)) {
            return next.getValue();
        }
    }

    return null;
}

public static OAuthNetworkManager createNetworkManagerAndConnect(InetAddress address, int serverPort,
        boolean useNativeTransport, OAuthCallback callback) {
    final OAuthNetworkManager networkmanager = new OAuthNetworkManager(EnumPacketDirection.CLIENTBOUND, callback);
    Class<? extends SocketChannel> oclass;
    LazyLoadBase<? extends EventLoopGroup> lazyloadbase;

    if (Epoll.isAvailable() && useNativeTransport) {
        oclass = EpollSocketChannel.class;
        lazyloadbase = CLIENT_EPOLL_EVENTLOOP;
    } else {
        oclass = NioSocketChannel.class;
        lazyloadbase = CLIENT_NIO_EVENTLOOP;
    }

    (new Bootstrap()).group(lazyloadbase.getValue()).handler(new ChannelInitializer<Channel>() {
        @Override
        protected void initChannel(Channel p_initChannel_1_) throws Exception {
            try {
                p_initChannel_1_.config().setOption(ChannelOption.TCP_NODELAY, Boolean.valueOf(true));
            } catch (ChannelException var3) {
                ;
            }

            p_initChannel_1_.pipeline().addLast("timeout", new ReadTimeoutHandler(30))
                    .addLast("splitter", new NettyVarint21FrameDecoder())
                    .addLast("decoder", new NettyPacketDecoder(EnumPacketDirection.CLIENTBOUND))
                    .addLast("prepender", new NettyVarint21FrameEncoder())
                    .addLast("encoder", new NettyPacketEncoder(EnumPacketDirection.SERVERBOUND))
                    .addLast("packet_handler", networkmanager);
        }
    }).channel(oclass).connect(address, serverPort).syncUninterruptibly();
    return networkmanager;
}

@Override
public void operationComplete(Future<Channel> cf) throws Exception {

    if (cf.isCancelled()) {
        promise.cancel(true);
        return;
    }

    if (!cf.isSuccess()) {
        promise.completeExceptionally(cf.cause());
        return;
    }

    Channel channel = cf.getNow();
    promise.whenComplete((result, error) -> pool.release(channel));

    try {

        FastdfsOperation<T> fastdfsOperation = new FastdfsOperation<>(channel, requester, replier, promise);
        if (LOG.isDebugEnabled()) {
            LOG.debug("execute {}", fastdfsOperation);
        }

        fastdfsOperation.execute();
    } catch (Exception e) {
        promise.completeExceptionally(e);
    }
}

/**
 * Filter Server register to broker every 10s ,if over 30s,no registration info.,remove it
 */
public void scanNotActiveChannel() {
    Iterator<Entry<Channel, FilterServerInfo>> it = this.filterServerTable.entrySet().iterator();
    while (it.hasNext()) {
        Entry<Channel, FilterServerInfo> next = it.next();
        long timestamp = next.getValue().getLastUpdateTimestamp();
        Channel channel = next.getKey();
        if ((System.currentTimeMillis() - timestamp) > FilterServerMaxIdleTimeMills) {
            log.info("The Filter Server<{}> expired, remove it", next.getKey());
            it.remove();
            RemotingUtil.closeChannel(channel);
        }
    }
}