Java源码示例:io.netty.handler.flush.FlushConsolidationHandler
示例1
@Override
protected void initChannel(Channel channel) throws Exception {
ChannelPipeline pipeline = channel.pipeline();
BoundNode node = channel.parent().attr(HANDLER).get();
node.clientChannelGroup.add(channel);
MDC.put("node", node.getId().toString());
try {
logger.debug("Got new connection {}", channel);
pipeline
.addLast(new FlushConsolidationHandler())
.addLast("decoder", new FrameDecoder(node.getFrameCodec()))
.addLast("encoder", new FrameEncoder(node.getFrameCodec()))
.addLast("requestHandler", new RequestHandler(node));
} finally {
MDC.remove("node");
}
}
示例2
@Override
public ChannelFuture bind(SocketAddress localAddress) {
ServerBootstrap boot = bootstrap();
initChannelFactory();
boot.childHandler(new ChannelInitializer<Channel>() {
@Override
protected void initChannel(Channel ch) throws Exception {
ch.pipeline().addLast(
new FlushConsolidationHandler(JConstants.EXPLICIT_FLUSH_AFTER_FLUSHES, true),
new IdleStateChecker(timer, JConstants.READER_IDLE_TIME_SECONDS, 0, 0),
idleStateTrigger,
CodecConfig.isCodecLowCopy() ? new LowCopyProtocolDecoder() : new ProtocolDecoder(),
encoder,
handler);
}
});
setOptions();
return boot.bind(localAddress);
}
示例3
@Override
public ChannelFuture bind(SocketAddress localAddress) {
ServerBootstrap boot = bootstrap();
initChannelFactory();
boot.childHandler(new ChannelInitializer<Channel>() {
@Override
protected void initChannel(Channel ch) throws Exception {
ch.pipeline().addLast(
new FlushConsolidationHandler(JConstants.EXPLICIT_FLUSH_AFTER_FLUSHES, true),
new IdleStateChecker(timer, JConstants.READER_IDLE_TIME_SECONDS, 0, 0),
idleStateTrigger,
CodecConfig.isCodecLowCopy() ? new LowCopyProtocolDecoder() : new ProtocolDecoder(),
encoder,
handler);
}
});
setOptions();
return boot.bind(localAddress);
}
示例4
@Override
public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress,
ChannelPromise promise) throws Exception {
// Remember the requested remote address for later use.
final InetSocketAddress inetRemoteAddr = (InetSocketAddress) remoteAddress;
this.remoteAddress = inetRemoteAddr;
// Configure the pipeline.
final Channel ch = ctx.channel();
ChannelUtil.disableWriterBufferWatermark(ch);
final ChannelPipeline p = ch.pipeline();
p.addLast(new FlushConsolidationHandler());
p.addLast(ReadSuppressingAndChannelDeactivatingHandler.INSTANCE);
try {
if (sslCtx != null) {
configureAsHttps(ch, inetRemoteAddr);
} else {
configureAsHttp(ch);
}
} catch (Throwable t) {
promise.tryFailure(t);
ctx.close();
} finally {
if (p.context(this) != null) {
p.remove(this);
}
}
ctx.connect(remoteAddress, localAddress, promise);
}
示例5
@Override
protected void initChannel(Channel ch) throws Exception {
ChannelUtil.disableWriterBufferWatermark(ch);
final ChannelPipeline p = ch.pipeline();
p.addLast(new FlushConsolidationHandler());
p.addLast(ReadSuppressingHandler.INSTANCE);
configurePipeline(p, port.protocols(), null);
}
示例6
@Override
protected void initChannel(SocketChannel ch) throws Exception {
OFChannelHandler handler = new OFChannelHandler(controller);
ChannelPipeline pipeline = ch.pipeline();
if (sslContext != null) {
log.info("OpenFlow SSL enabled.");
SSLEngine sslEngine = sslContext.createSSLEngine();
sslEngine.setNeedClientAuth(true);
sslEngine.setUseClientMode(false);
sslEngine.setEnabledProtocols(sslEngine.getSupportedProtocols());
sslEngine.setEnabledCipherSuites(sslEngine.getSupportedCipherSuites());
sslEngine.setEnableSessionCreation(true);
SslHandler sslHandler = new SslHandler(sslEngine);
pipeline.addLast("ssl", sslHandler);
} else {
log.debug("OpenFlow SSL disabled.");
}
pipeline.addLast("ofmessageencoder", OFMessageEncoder.getInstance());
pipeline.addLast("ofmessagedecoder", OFMessageDecoder.getInstance());
pipeline.addLast("consolidateflush", new FlushConsolidationHandler(
FlushConsolidationHandler.DEFAULT_EXPLICIT_FLUSH_AFTER_FLUSHES, true));
pipeline.addLast("idle", new IdleStateHandler(5, 25, 0));
pipeline.addLast("timeout", new ReadTimeoutHandler(30));
// XXX S ONOS: was 15 increased it to fix Issue #296
pipeline.addLast("handshaketimeout",
new HandshakeTimeoutHandler(handler, 60));
// ExecutionHandler equivalent now part of Netty core
if (pipelineExecutor != null) {
pipeline.addLast(pipelineExecutor, "handler", handler);
} else {
pipeline.addLast("handler", handler);
}
}
示例7
@Override
public JConnection connect(UnresolvedAddress address, boolean async) {
setOptions();
final Bootstrap boot = bootstrap();
final SocketAddress socketAddress = InetSocketAddress.createUnresolved(address.getHost(), address.getPort());
final JChannelGroup group = group(address);
// 重连watchdog
final ConnectionWatchdog watchdog = new ConnectionWatchdog(boot, timer, socketAddress, group) {
@Override
public ChannelHandler[] handlers() {
return new ChannelHandler[] {
new FlushConsolidationHandler(JConstants.EXPLICIT_FLUSH_AFTER_FLUSHES, true),
this,
new IdleStateChecker(timer, 0, JConstants.WRITER_IDLE_TIME_SECONDS, 0),
idleStateTrigger,
CodecConfig.isCodecLowCopy() ? new LowCopyProtocolDecoder() : new ProtocolDecoder(),
encoder,
handler
};
}
};
ChannelFuture future;
try {
synchronized (bootstrapLock()) {
boot.handler(new ChannelInitializer<Channel>() {
@Override
protected void initChannel(Channel ch) throws Exception {
ch.pipeline().addLast(watchdog.handlers());
}
});
future = boot.connect(socketAddress);
}
// 以下代码在synchronized同步块外面是安全的
if (!async) {
future.sync();
}
} catch (Throwable t) {
throw new ConnectFailedException("Connects to [" + address + "] fails", t);
}
return new JNettyConnection(address, future) {
@Override
public void setReconnect(boolean reconnect) {
if (reconnect) {
watchdog.start();
} else {
watchdog.stop();
}
}
};
}
示例8
@Override
public JConnection connect(UnresolvedAddress address, boolean async) {
setOptions();
final Bootstrap boot = bootstrap();
final SocketAddress socketAddress = new DomainSocketAddress(address.getPath());
final JChannelGroup group = group(address);
// 重连watchdog
final ConnectionWatchdog watchdog = new ConnectionWatchdog(boot, timer, socketAddress, group) {
@Override
public ChannelHandler[] handlers() {
return new ChannelHandler[] {
new FlushConsolidationHandler(JConstants.EXPLICIT_FLUSH_AFTER_FLUSHES, true),
this,
new IdleStateChecker(timer, 0, JConstants.WRITER_IDLE_TIME_SECONDS, 0),
idleStateTrigger,
CodecConfig.isCodecLowCopy() ? new LowCopyProtocolDecoder() : new ProtocolDecoder(),
encoder,
handler
};
}
};
ChannelFuture future;
try {
synchronized (bootstrapLock()) {
boot.handler(new ChannelInitializer<Channel>() {
@Override
protected void initChannel(Channel ch) throws Exception {
ch.pipeline().addLast(watchdog.handlers());
}
});
future = boot.connect(socketAddress);
}
// 以下代码在synchronized同步块外面是安全的
if (!async) {
future.sync();
}
} catch (Throwable t) {
throw new ConnectFailedException("Connects to [" + address + "] fails", t);
}
return new JNettyConnection(address, future) {
@Override
public void setReconnect(boolean reconnect) {
if (reconnect) {
watchdog.start();
} else {
watchdog.stop();
}
}
};
}