io.netty.channel.ChannelHandlerContext#channel ( )源码实例Demo

@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    Channel channel = ctx.channel();
    dataCenterChannelStore.isDcChannelToSave(channel);
    if (IdleStateEvent.class.isAssignableFrom(evt.getClass())) {
        IdleStateEvent event = (IdleStateEvent) evt;
        if (event.state() == IdleState.ALL_IDLE) {
            //发送心跳
            channel.writeAndFlush(PING);
        }
        if (event.state() == IdleState.READER_IDLE) {
            //发送心跳
            channel.writeAndFlush(PING);
        }
        if (event.state() == IdleState.WRITER_IDLE) {
            channel.writeAndFlush(PING);
        }
    } else {
        super.userEventTriggered(ctx, evt);
    }
}
 

@Override
public void channelActive(final ChannelHandlerContext ctx) {
    UkcpChannel kcpCh = (UkcpChannel) ctx.channel();
    kcpCh.conv(KcpRttClient.CONV); // set conv

    future = scheduleSrv.scheduleWithFixedDelay(new Runnable() {
        @Override
        public void run() {
            ctx.write(rttMsg(++count));
            if (count >= rtts.length) {
                // finish
                future.cancel(true);
                ctx.write(rttMsg(-1));

            }
            ctx.flush();
        }
    }, KcpRttClient.RTT_INTERVAL, KcpRttClient.RTT_INTERVAL, TimeUnit.MILLISECONDS);
}
 

@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object event) throws Exception {
    if (event instanceof ConnectionEventType) {
        switch ((ConnectionEventType) event) {
            case CONNECT:
                Channel channel = ctx.channel();
                if (null != channel) {
                    Connection connection = channel.attr(Connection.CONNECTION).get();
                    this.onEvent(connection, connection.getUrl().getOriginUrl(),
                        ConnectionEventType.CONNECT);
                } else {
                    logger
                        .warn("channel null when handle user triggered event in ConnectionEventHandler!");
                }
                break;
            default:
                break;
        }
    } else {
        super.userEventTriggered(ctx, event);
    }
}
 

@Override
public void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
    // Send the received message to all channels but the current one.
    for (Channel c: channels) {
        if (c != ctx.channel()) {
            c.writeAndFlush("[" + ctx.channel().remoteAddress() + "] " + msg + '\n');
        } else {
            c.writeAndFlush("[you] " + msg + '\n');
        }
    }

    // Close the connection if the client has sent 'bye'.
    if ("bye".equals(msg.toLowerCase())) {
        ctx.close();
    }
}
 

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
	if(ctx.channel()!= null && ctx.channel().remoteAddress()!=null){
		LOG.error("LionServerChannelInboundHanndler happen exception.remote address "+ctx.channel().remoteAddress().toString(),cause);
	}
    if(ctx != null){
        ctx.close();
    }
}
 

@Override
public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
    super.channelUnregistered(ctx);
    SocketChannel channel = (SocketChannel) ctx.channel();
    String nodeId = IpUtil.getNodeId(channel.remoteAddress());
    Attribute<Node> nodeAttribute = channel.attr(AttributeKey.valueOf("node-" + nodeId));

    Node node = nodeAttribute.get();
    if (node != null && node.getDisconnectListener() != null) {
        node.getDisconnectListener().action();
    }
    LoggerUtil.COMMON_LOG.info("Server Node is channelUnregistered:{}:{}", channel.remoteAddress().getHostString(), channel.remoteAddress().getPort());
}
 

/**
 * 关闭用户连接
 */
private void closeChannle(ChannelHandlerContext ctx) {

    Channel channel = ctx.channel();

    if (channel != null && channel.isActive()) {
        channel.close();
    }
}
 

@Override
public void channelInactive(ChannelHandlerContext context) throws Exception {
    Channel channel = context.channel();
    InetSocketAddress address = InetSocketAddress.class.cast(channel.remoteAddress());
    CommunicationSession<Channel> session = sessionManager.getSession(address);
    // 将会话放到定时队列
    Instant now = Instant.now();
    Instant expire = now.plusMillis(expired);
    DelayElement<CommunicationSession<Channel>> element = new DelayElement<>(session, expire);
    queue.offer(element);
    super.channelInactive(context);
}
 

@Override
public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause) throws Exception {

    final Channel channel = ctx.channel();

    if (cause instanceof SSLException) {
        //We can ignore SSL Exceptions, since the channel gets closed anyway.
        return;

    } else if (cause instanceof ClosedChannelException) {
        //We can ignore this because the channel is already closed
        return;

    } else if (cause instanceof IOException) {

        //We can ignore this because the channel is already closed because of an IO problem
        return;

    } else if (cause instanceof CorruptedFrameException) {

        //We can ignore this because the channel is already closed because of an IO problem
        eventLog.clientWasDisconnected(channel, "Illegal websocket data sent by client: " + cause.getMessage());
        channel.close();
        return;


    } else if (cause instanceof IllegalArgumentException) {

        //do not log IllegalArgumentException as error

    } else {
        log.error("An unexpected error occurred for client with IP {}: {}",
                ChannelUtils.getChannelIP(channel).or("UNKNOWN"), ExceptionUtils.getStackTrace(cause));
    }

    if (channel != null) {
        eventLog.clientWasDisconnected(channel, "Channel exception: " + cause.getMessage());
        channel.close();
    }
}
 

@Override
protected void channelRead0(ChannelHandlerContext context, String string) throws Exception {
	final Channel channel = context.channel();
	String response = this.generateResponse(string.toUpperCase());
	if(response != null) {
		channel.writeAndFlush(response).addListener(new ChannelFutureListener() {
			public void operationComplete(ChannelFuture future) throws Exception {
				channel.close();
			}
		});
	} else {
		channel.close();
	}
}
 

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    channel = ctx.channel();
    if (!autoRead) {
        ctx.read();
    }
}
 

@Override
public void channelInactive(ChannelHandlerContext context) throws Exception {
  Channel channel = context.channel();
  NettyConnection connection = removeConnection(channel);
  if (connection != null) {
    connection.handleClosed();
  }
}
 

@Override
public void channelActive(ChannelHandlerContext ctx) {
    clientChannel = ctx.channel();
}
 

@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
    // 每当服务端收到客户端断开连接的时候
    Channel channel = ctx.channel();
    channel.writeAndFlush("client - " + channel.remoteAddress() + "退出了聊天室\n");
}
 

@Override
protected void channelRead0(@Nullable ChannelHandlerContext ctx, @Nullable Object msg) throws Exception {
   if (ctx == null || msg == null) {
      return;
   }

   lastPlatformMsg = System.nanoTime();
   Channel ch = ctx.channel();
   if (!handshaker.isHandshakeComplete()) {
      handshaker.finishHandshake(ch, (FullHttpResponse) msg);

      connected = true;
      handshakeFuture.setSuccess();
      return;
   }

   if (msg instanceof FullHttpResponse) {
      log.warn("unxpected full http response: {}", msg);
      ctx.close();
      return;
   }

   WebSocketFrame frame = (WebSocketFrame) msg;
   if (frame instanceof BinaryWebSocketFrame) {
      websocketFrameBuf.clear();
      websocketFrameBuf.writeBytes(frame.content());
   } else if (frame instanceof ContinuationWebSocketFrame){
      if (websocketFrameBuf.isReadable()) {
         websocketFrameBuf.writeBytes(frame.content());
      } else {
         log.warn("continuation frame received without initial frame.");
         ctx.close();
      }
   } else if (frame instanceof PingWebSocketFrame) {
      log.trace("received websocket ping request from platform");
      ctx.writeAndFlush(new PongWebSocketFrame(frame.content().retain()));
      lastHubMsg = System.nanoTime();
      return;
   } else if (frame instanceof PongWebSocketFrame) {
      log.trace("received websocket pong response from platform");
      return;
   } else if (frame instanceof CloseWebSocketFrame) {
      log.warn("received websocket close request");
      ctx.close();
      return;
   }

   if (frame.isFinalFragment()) {
      decodeHubFrame(ctx, websocketFrameBuf);
   }
}
 

@Override
public void channelActive(ChannelHandlerContext ctx)
        throws Exception {
    channel = ctx.channel();
}
 

@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
    if(ctx.channel() == this.channel) {
        this.disconnect("Connection closed.");
    }
}
 

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    channel = ctx.channel();
}
 

public static String channelLongText(ChannelHandlerContext ctx) {
    Channel channel = ctx.channel();
    return channel.id().asLongText();
}
 

/**
 * 判断session是否存在
 * @param context
 * @return
 */
public boolean containsSession(ChannelHandlerContext context){
    return context!=null && context.channel()!=null && context.channel().id()!=null && manager.sessionMap.get(context.channel().id())!=null;
}