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

@Override
public void channelActive(ChannelHandlerContext ctx) {
	Channel channel = ctx.channel();
	if (channel.isActive()) {
		if (ctx.pipeline().get(NettyPipeline.H2MultiplexHandler) == null) {
			// Proceed with HTTP/1.x as per configuration
			ctx.fireChannelActive();
		}
		else if (ctx.pipeline().get(NettyPipeline.SslHandler) == null) {
			// Proceed with H2C as per configuration
			sendNewState(Connection.from(channel), ConnectionObserver.State.CONNECTED);
			ctx.flush();
			ctx.read();
		}
		else {
			// Proceed with H2 as per configuration
			sendNewState(Connection.from(channel), ConnectionObserver.State.CONNECTED);
		}
	}
}
 

public static void main(String[] args) throws Exception {
	InputStream ins=TestWssClient.class.getResourceAsStream("cloud.jueb.net.pfx");
	String strPassword="xxxxxx";
	SslContext sslc=NettyServerSslUtil.buildSslContext_P12_Pfx(ins, strPassword);
	NettyServerConfig nc=new NettyServerConfig();
	NettyServer ns=new NettyServer(nc, "0.0.0.0", 1191,new WebSocketServerInitializer("/test",sslc) {
		@Override
		protected void webSocketHandComplete(ChannelHandlerContext ctx) {
			ChannelPipeline p=ctx.pipeline();
			p.addLast(new WebSocketTextFrameStringAdapter());//消息解码器
			p.addLast(new DefaultIdleListenerHandler<String>(new Listener()));//心跳适配器
			//为新加的handler手动触发必要事件
			ctx.fireChannelRegistered();
			ctx.fireChannelActive();
		}
	});
	ns.start();
	new Scanner(System.in).nextLine();
}
 

/**
 * Helper method to walk the iterator and create a new request that defines which control param
 * should be negotiated right now.
 */
private void negotiate(final ChannelHandlerContext ctx) {
  if (controlSettings.hasNext()) {
    Map.Entry<String, String> setting = controlSettings.next();

    LOGGER.debug("Negotiating DCP Control {}: {}", setting.getKey(), setting.getValue());
    ByteBuf request = ctx.alloc().buffer();
    DcpControlRequest.init(request);
    DcpControlRequest.key(setting.getKey(), request);
    DcpControlRequest.value(Unpooled.copiedBuffer(setting.getValue(), UTF_8), request);

    ctx.writeAndFlush(request);
  } else {
    originalPromise().setSuccess();
    ctx.pipeline().remove(this);
    ctx.fireChannelActive();
    LOGGER.debug("Negotiated all DCP Control settings against Node {}", ctx.channel().remoteAddress());
  }
}
 

@Override
public void channelActive(ChannelHandlerContext ctx) {
    DefaultFullHttpRequest upgradeRequest =
            new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/");
    ctx.writeAndFlush(upgradeRequest);

    ctx.fireChannelActive();

    // Done with this handler, remove it from the pipeline.
    ctx.pipeline().remove(this);

    configureEndOfPipeline(ctx.pipeline());
}
 

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    if (state == INACTIVE) {
        state = RUNNING;
        maybeRequestMore();
    }
    ctx.fireChannelActive();
}
 

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    FullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/");
    request.headers().set(HttpHeaderNames.HOST, host);
    request.headers().set(HttpHeaderNames.USER_AGENT, "netty-ocsp-example/1.0");

    ctx.writeAndFlush(request).addListener(ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE);

    ctx.fireChannelActive();
}
 

/**
 * 服务端监听到客户端活动
 * 
 * @param ctx 连接context
 * @throws Exception
 */
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    // 服务端接收到客户端上线通知
    Channel incoming = ctx.channel();
    logger.debug("MessageServerHandler:" + incoming.remoteAddress() + "在线");
    login(ctx);
    ctx.fireChannelActive();
}
 

@Override
protected void webSocketHandComplete(ChannelHandlerContext ctx) {
	ctx.channel().pipeline().addLast(new WebSocketTextFrameStringAdapter());//适配器
	ctx.channel().pipeline().addLast(handler);
	//为新加的handler手动触发必要事件
	ctx.fireChannelRegistered();
	ctx.fireChannelActive();
}
 

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
	if (ChannelUtil.clientSide(ctx)) {
		negotiate(ctx);
	}
	ctx.fireChannelActive();
}
 

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
  String remoteAddress =
      ((InetSocketAddress) ctx.channel().remoteAddress()).getAddress().getHostAddress();

  if (blackList.contains(remoteAddress)) {
    ctx.channel().attr(XrpcConstants.IP_BLACK_LIST).set(Boolean.TRUE);
  }

  ctx.fireChannelActive();
}
 

public void test(final boolean dropChannelActive) throws Exception {
     SSLEngine engine = SSLContext.getDefault().createSSLEngine();
     engine.setUseClientMode(true);

     EmbeddedChannel ch = new EmbeddedChannel(false, false,
             this,
             new SslHandler(engine),
             new ChannelInboundHandlerAdapter() {
                 @Override
                 public void channelActive(ChannelHandlerContext ctx) throws Exception {
                     if (!dropChannelActive) {
                         ctx.fireChannelActive();
                     }
                 }
             }
     );
     ch.config().setAutoRead(false);
     assertFalse(ch.config().isAutoRead());

     ch.register();

     assertTrue(readIssued);
     readIssued = false;

     assertTrue(ch.writeOutbound(Unpooled.EMPTY_BUFFER));
     assertTrue(readIssued);
     assertTrue(ch.finishAndReleaseAll());
}
 

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    // If we subscribed before the channel was active, then our read would have been ignored.
    if (state == DEMANDING) {
        requestDemand();
    }
    ctx.fireChannelActive();
}
 

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
  if (noOp) {
    ctx.pipeline().remove(this);
    ctx.fireChannelActive();
  }
}
 

@Override
public void channelActive(ChannelHandlerContext ctx) {
    ctx.pipeline().get(SslHandler.class).handshakeFuture().addListener(
            (GenericFutureListener<Future<Channel>>) future -> {
                if (future.isSuccess()) {
                    logger.info("Ssl handshake is successed, session is protected by " + ctx.pipeline().get(SslHandler.class).engine().getSession().getCipherSuite());
                } else {
                    ctx.channel().close();
                }
            });
    ctx.fireChannelActive();
}
 

/**
 * 建立连接时
 */
@Override  
public void channelActive(ChannelHandlerContext ctx) throws Exception {  
    System.out.println("激活时间是："+MyTools.getNowTime(""));  
    ctx.fireChannelActive();  
}
 

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
  ctx.pipeline().remove(this);
  ctx.fireChannelActive();
}
 

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

/**
 * The {@link Channel} of the {@link ChannelHandlerContext} is now active
 *
 * @param context {@link ChannelHandlerContext} to which this {@link RntbdRequestManager} belongs
 */
@Override
public void channelActive(final ChannelHandlerContext context) {
    this.traceOperation(context, "channelActive");
    context.fireChannelActive();
}
 

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
  if (DEBUG_CALLS) Gdx.app.debug(TAG, "channelActive");
  ctx.fireChannelActive();
}
 

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    eventPublisher.offer(new ActiveEvent(channel));
    ctx.fireChannelActive();
}