下面列出了怎么用io.netty.channel.local.LocalEventLoopGroup的API类实例代码及写法,或者点击链接到github查看源代码。
/**
* Tests that if channel was unhealthy it is not offered back to the pool.
*
* @throws Exception
*/
@Test
public void testUnhealthyChannelIsNotOffered() throws Exception {
EventLoopGroup group = new LocalEventLoopGroup();
LocalAddress addr = new LocalAddress(LOCAL_ADDR_ID);
Bootstrap cb = new Bootstrap();
cb.remoteAddress(addr);
cb.group(group)
.channel(LocalChannel.class);
ServerBootstrap sb = new ServerBootstrap();
sb.group(group)
.channel(LocalServerChannel.class)
.childHandler(new ChannelInitializer<LocalChannel>() {
@Override
public void initChannel(LocalChannel ch) throws Exception {
ch.pipeline().addLast(new ChannelInboundHandlerAdapter());
}
});
// Start server
Channel sc = sb.bind(addr).syncUninterruptibly().channel();
ChannelPoolHandler handler = new CountingChannelPoolHandler();
ChannelPool pool = new SimpleChannelPool(cb, handler);
Channel channel1 = pool.acquire().syncUninterruptibly().getNow();
pool.release(channel1).syncUninterruptibly();
Channel channel2 = pool.acquire().syncUninterruptibly().getNow();
//first check that when returned healthy then it actually offered back to the pool.
assertSame(channel1, channel2);
channel1.close().syncUninterruptibly();
pool.release(channel1).syncUninterruptibly();
Channel channel3 = pool.acquire().syncUninterruptibly().getNow();
//channel1 was not healthy anymore so it should not get acquired anymore.
assertNotSame(channel1, channel3);
sc.close().syncUninterruptibly();
channel3.close().syncUninterruptibly();
group.shutdownGracefully();
}
/**
* Tests that if channel was unhealthy it is was offered back to the pool because
* it was requested not to validate channel health on release.
*
* @throws Exception
*/
@Test
public void testUnhealthyChannelIsOfferedWhenNoHealthCheckRequested() throws Exception {
EventLoopGroup group = new LocalEventLoopGroup();
LocalAddress addr = new LocalAddress(LOCAL_ADDR_ID);
Bootstrap cb = new Bootstrap();
cb.remoteAddress(addr);
cb.group(group)
.channel(LocalChannel.class);
ServerBootstrap sb = new ServerBootstrap();
sb.group(group)
.channel(LocalServerChannel.class)
.childHandler(new ChannelInitializer<LocalChannel>() {
@Override
public void initChannel(LocalChannel ch) throws Exception {
ch.pipeline().addLast(new ChannelInboundHandlerAdapter());
}
});
// Start server
Channel sc = sb.bind(addr).syncUninterruptibly().channel();
ChannelPoolHandler handler = new CountingChannelPoolHandler();
ChannelPool pool = new SimpleChannelPool(cb, handler, ChannelHealthChecker.ACTIVE, false);
Channel channel1 = pool.acquire().syncUninterruptibly().getNow();
channel1.close().syncUninterruptibly();
Future<Void> releaseFuture =
pool.release(channel1, channel1.eventLoop().<Void>newPromise()).syncUninterruptibly();
assertThat(releaseFuture.isSuccess(), CoreMatchers.is(true));
Channel channel2 = pool.acquire().syncUninterruptibly().getNow();
//verifying that in fact the channel2 is different that means is not pulled from the pool
assertNotSame(channel1, channel2);
sc.close().syncUninterruptibly();
channel2.close().syncUninterruptibly();
group.shutdownGracefully();
}
@Test(expected = ConnectException.class)
public void testMap() throws Exception {
EventLoopGroup group = new LocalEventLoopGroup();
LocalAddress addr = new LocalAddress(LOCAL_ADDR_ID);
final Bootstrap cb = new Bootstrap();
cb.remoteAddress(addr);
cb.group(group)
.channel(LocalChannel.class);
AbstractChannelPoolMap<EventLoop, SimpleChannelPool> poolMap =
new AbstractChannelPoolMap<EventLoop, SimpleChannelPool>() {
@Override
protected SimpleChannelPool newPool(EventLoop key) {
return new SimpleChannelPool(cb.clone(key), new TestChannelPoolHandler());
}
};
EventLoop loop = group.next();
assertFalse(poolMap.iterator().hasNext());
assertEquals(0, poolMap.size());
SimpleChannelPool pool = poolMap.get(loop);
assertEquals(1, poolMap.size());
assertTrue(poolMap.iterator().hasNext());
assertSame(pool, poolMap.get(loop));
assertTrue(poolMap.remove(loop));
assertFalse(poolMap.remove(loop));
assertFalse(poolMap.iterator().hasNext());
assertEquals(0, poolMap.size());
pool.acquire().syncUninterruptibly();
}
@Test(timeout = 5000)
public void testHandlerRegister() throws Exception {
final CountDownLatch latch = new CountDownLatch(1);
final AtomicReference<Throwable> error = new AtomicReference<Throwable>();
LocalEventLoopGroup group = new LocalEventLoopGroup(1);
try {
ServerBootstrap sb = new ServerBootstrap();
sb.channel(LocalServerChannel.class)
.group(group)
.childHandler(new ChannelInboundHandlerAdapter())
.handler(new ChannelHandlerAdapter() {
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
try {
assertTrue(ctx.executor().inEventLoop());
} catch (Throwable cause) {
error.set(cause);
} finally {
latch.countDown();
}
}
});
sb.register().syncUninterruptibly();
latch.await();
assertNull(error.get());
} finally {
group.shutdownGracefully();
}
}
ServerBootstrap getLocalServerBootstrap() {
EventLoopGroup serverGroup = new LocalEventLoopGroup();
ServerBootstrap sb = new ServerBootstrap();
sb.group(serverGroup);
sb.channel(LocalServerChannel.class);
sb.childHandler(new ChannelInitializer<LocalChannel>() {
@Override
public void initChannel(LocalChannel ch) throws Exception {
}
});
return sb;
}
Bootstrap getLocalClientBootstrap() {
EventLoopGroup clientGroup = new LocalEventLoopGroup();
Bootstrap cb = new Bootstrap();
cb.channel(LocalChannel.class);
cb.group(clientGroup);
cb.handler(loggingHandler);
return cb;
}
@Test
public void shouldShowUnmanagedCustomResourcesInEnvDump() {
//create an environment with a custom IOPool and Scheduler that are not cleaned up on shutdown
env = DefaultCoreEnvironment.builder()
.ioPool(new LocalEventLoopGroup())
.scheduler(Schedulers.newThread()).build();
String dump = env.dumpParameters(new StringBuilder()).toString();
assertTrue(dump, dump.contains("LocalEventLoopGroup!unmanaged"));
assertTrue(dump, dump.contains("NewThreadScheduler!unmanaged"));
}
@BeforeClass
public static void createEventLoop() {
group = new LocalEventLoopGroup();
}
@Test(timeout = 10000)
public void testBindDeadLock() throws Exception {
EventLoopGroup groupA = new LocalEventLoopGroup(1);
EventLoopGroup groupB = new LocalEventLoopGroup(1);
try {
ChannelInboundHandler dummyHandler = new DummyHandler();
final Bootstrap bootstrapA = new Bootstrap();
bootstrapA.group(groupA);
bootstrapA.channel(LocalChannel.class);
bootstrapA.handler(dummyHandler);
final Bootstrap bootstrapB = new Bootstrap();
bootstrapB.group(groupB);
bootstrapB.channel(LocalChannel.class);
bootstrapB.handler(dummyHandler);
List<Future<?>> bindFutures = new ArrayList<Future<?>>();
// Try to bind from each other.
for (int i = 0; i < 1024; i ++) {
bindFutures.add(groupA.next().submit(new Runnable() {
@Override
public void run() {
bootstrapB.bind(LocalAddress.ANY);
}
}));
bindFutures.add(groupB.next().submit(new Runnable() {
@Override
public void run() {
bootstrapA.bind(LocalAddress.ANY);
}
}));
}
for (Future<?> f: bindFutures) {
f.sync();
}
} finally {
groupA.shutdownGracefully();
groupB.shutdownGracefully();
groupA.terminationFuture().sync();
groupB.terminationFuture().sync();
}
}
@Test(timeout = 10000)
public void testConnectDeadLock() throws Exception {
EventLoopGroup groupA = new LocalEventLoopGroup(1);
EventLoopGroup groupB = new LocalEventLoopGroup(1);
try {
ChannelInboundHandler dummyHandler = new DummyHandler();
final Bootstrap bootstrapA = new Bootstrap();
bootstrapA.group(groupA);
bootstrapA.channel(LocalChannel.class);
bootstrapA.handler(dummyHandler);
final Bootstrap bootstrapB = new Bootstrap();
bootstrapB.group(groupB);
bootstrapB.channel(LocalChannel.class);
bootstrapB.handler(dummyHandler);
List<Future<?>> bindFutures = new ArrayList<Future<?>>();
// Try to connect from each other.
for (int i = 0; i < 1024; i ++) {
bindFutures.add(groupA.next().submit(new Runnable() {
@Override
public void run() {
bootstrapB.connect(LocalAddress.ANY);
}
}));
bindFutures.add(groupB.next().submit(new Runnable() {
@Override
public void run() {
bootstrapA.connect(LocalAddress.ANY);
}
}));
}
for (Future<?> f: bindFutures) {
f.sync();
}
} finally {
groupA.shutdownGracefully();
groupB.shutdownGracefully();
groupA.terminationFuture().sync();
groupB.terminationFuture().sync();
}
}