IpfixCollectorClientInitializer extends ChannelInitializer<SocketChannel> { @Override protected void initChannel(SocketChannel socketChannel) throws Exception { LOGGER.debug("Initializing channel"); socketChannel.config().setKeepAlive(false); socketChannel.closeFuture().addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture channelFuture) throws Exception { m_clientConnectedCount.decrementAndGet(); LOGGER.debug(String.format("Channel {} is closed", channelFuture.channel())); } }); if (m_clientConnectedCount.get() >= getIpfixCollectorMaxConnection()) { LOGGER.debug(String.format("Current tcp connection is %s", m_clientConnectedCount.get())); LOGGER.info(String.format("Max tcp connection is reached: %s", IPFIX_COLLECTOR_MAX_CONNECTION)); socketChannel.close(); return; } m_clientConnectedCount.incrementAndGet(); LOGGER.debug(String.format("Current number of tcp connection is %s", m_clientConnectedCount.get())); IpfixCollectorTimeoutHandler timeoutHandler = createTimeoutHandler(); socketChannel.pipeline() .addLast(new IpfixMessageDecoder()) .addLast(timeoutHandler) .addLast(new IpfixCollectorHandler(m_collectingService)); } IpfixCollectorClientInitializer(CollectingService collectingService, AtomicLong clientConnectedCount); @Override void exceptionCaught(ChannelHandlerContext ctx, Throwable cause);  }
@Test public void testInitChannel() throws Exception { for (int i = 0; i < INCOMING_CONNECTION; i++) { IpfixCollectorClientInitializer client = createClient(); SocketChannel chanel = createChanel(); client.initChannel(chanel); } assertEquals(MAX_CONNECTION, m_clientConnectedCount.get()); }
@Test public void testInitChannelWithTimeoutHandler() throws Exception { IpfixCollectorTimeoutHandler timeoutHandler = mock(IpfixCollectorTimeoutHandler.class); IpfixCollectorClientInitializer client = new IpfixCollectorClientInitializer(m_collectingService, m_clientConnectedCount) { @Override protected IpfixCollectorTimeoutHandler createTimeoutHandler() { return timeoutHandler; } }; ChannelHandlerContext context = mock(ChannelHandlerContext.class); doAnswer(invocationOnMock -> { ChannelHandlerContext channelHandlerContext = (ChannelHandlerContext) invocationOnMock.getArguments()[0]; channelHandlerContext.close(); return null; }).when(timeoutHandler).readTimedOut(any()); SocketChannel channel = createChannelWithMockTimeoutHandler(context); client.initChannel(channel); verify(context).close(); }