@Test public void testCreateClient() throws Exception { HandlerFactory handlerFactory = HandlerFactory.instanceAddDecoder(MessageDecoder.createClientProfile(sxpNode), HandlerFactory.Position.END); SxpConnection connection = mock(SxpConnection.class); when(connection.getPassword()).thenReturn("passwd"); when(connection.getDestination()).thenReturn(new InetSocketAddress("0.0.0.0", 64999)); when(connection.getSecurityType()).thenReturn(SecurityType.Default); Channel channel = ConnectFacade.createClient(sxpNode, connection, handlerFactory).channel(); assertNotNull(channel.config().getAllocator()); assertTrue(channel.isOpen()); assertTrue(channel.isWritable()); channel.close().get(); when(connection.getSecurityType()).thenReturn(SecurityType.TLS); channel = ConnectFacade.createClient(sxpNode, connection, handlerFactory).channel(); assertNotNull(channel.config().getAllocator()); assertTrue(channel.isOpen()); assertTrue(channel.isWritable()); channel.close().get(); }
public static ChannelFuture createClient(SxpNode node, SxpConnection connection, HandlerFactory hf) { if (!Epoll.isAvailable()) { throw new UnsupportedOperationException(Epoll.unavailabilityCause().getCause()); } SecurityType connectionSecurityType = connection.getSecurityType(); Optional<SslContext> clientSslContext = node.getSslContextFactory().getClientContext(); RecvByteBufAllocator recvByteBufAllocator = new FixedRecvByteBufAllocator(Constants.MESSAGE_LENGTH_MAX); Bootstrap bootstrap = new Bootstrap(); if (connection.hasNonEmptyPassword()) { bootstrap.option(EpollChannelOption.TCP_MD5SIG, Collections.singletonMap(connection.getDestination().getAddress(), connection.getPassword().getBytes(StandardCharsets.US_ASCII))); } else if ((connectionSecurityType == SecurityType.TLS) && !clientSslContext.isPresent()) { throw new IllegalStateException( String.format("%s has TSL enabled but %s does not provide any certificates.", connection, node)); } bootstrap.channel(EpollSocketChannel.class) .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, Configuration.NETTY_CONNECT_TIMEOUT_MILLIS) .option(ChannelOption.RCVBUF_ALLOCATOR, recvByteBufAllocator) .option(ChannelOption.TCP_NODELAY, true) .option(ChannelOption.SO_KEEPALIVE, true) .localAddress(node.getSourceIp().getHostAddress(), 0) .group(EVENT_LOOP_GROUP) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { if ((connectionSecurityType == SecurityType.TLS)) { ch.pipeline().addLast(clientSslContext.get().newHandler(ch.alloc())); } ch.pipeline().addLast(hf.getDecoders()); ch.pipeline().addLast(hf.getEncoders()); } }); return bootstrap.connect(connection.getDestination()); }
ConnectFacade { public static ChannelFuture createClient(SxpNode node, SxpConnection connection, HandlerFactory hf) { if (!Epoll.isAvailable()) { throw new UnsupportedOperationException(Epoll.unavailabilityCause().getCause()); } SecurityType connectionSecurityType = connection.getSecurityType(); Optional<SslContext> clientSslContext = node.getSslContextFactory().getClientContext(); RecvByteBufAllocator recvByteBufAllocator = new FixedRecvByteBufAllocator(Constants.MESSAGE_LENGTH_MAX); Bootstrap bootstrap = new Bootstrap(); if (connection.hasNonEmptyPassword()) { bootstrap.option(EpollChannelOption.TCP_MD5SIG, Collections.singletonMap(connection.getDestination().getAddress(), connection.getPassword().getBytes(StandardCharsets.US_ASCII))); } else if ((connectionSecurityType == SecurityType.TLS) && !clientSslContext.isPresent()) { throw new IllegalStateException( String.format("%s has TSL enabled but %s does not provide any certificates.", connection, node)); } bootstrap.channel(EpollSocketChannel.class) .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, Configuration.NETTY_CONNECT_TIMEOUT_MILLIS) .option(ChannelOption.RCVBUF_ALLOCATOR, recvByteBufAllocator) .option(ChannelOption.TCP_NODELAY, true) .option(ChannelOption.SO_KEEPALIVE, true) .localAddress(node.getSourceIp().getHostAddress(), 0) .group(EVENT_LOOP_GROUP) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { if ((connectionSecurityType == SecurityType.TLS)) { ch.pipeline().addLast(clientSslContext.get().newHandler(ch.alloc())); } ch.pipeline().addLast(hf.getDecoders()); ch.pipeline().addLast(hf.getEncoders()); } }); return bootstrap.connect(connection.getDestination()); } }
ConnectFacade { public static ChannelFuture createClient(SxpNode node, SxpConnection connection, HandlerFactory hf) { if (!Epoll.isAvailable()) { throw new UnsupportedOperationException(Epoll.unavailabilityCause().getCause()); } SecurityType connectionSecurityType = connection.getSecurityType(); Optional<SslContext> clientSslContext = node.getSslContextFactory().getClientContext(); RecvByteBufAllocator recvByteBufAllocator = new FixedRecvByteBufAllocator(Constants.MESSAGE_LENGTH_MAX); Bootstrap bootstrap = new Bootstrap(); if (connection.hasNonEmptyPassword()) { bootstrap.option(EpollChannelOption.TCP_MD5SIG, Collections.singletonMap(connection.getDestination().getAddress(), connection.getPassword().getBytes(StandardCharsets.US_ASCII))); } else if ((connectionSecurityType == SecurityType.TLS) && !clientSslContext.isPresent()) { throw new IllegalStateException( String.format("%s has TSL enabled but %s does not provide any certificates.", connection, node)); } bootstrap.channel(EpollSocketChannel.class) .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, Configuration.NETTY_CONNECT_TIMEOUT_MILLIS) .option(ChannelOption.RCVBUF_ALLOCATOR, recvByteBufAllocator) .option(ChannelOption.TCP_NODELAY, true) .option(ChannelOption.SO_KEEPALIVE, true) .localAddress(node.getSourceIp().getHostAddress(), 0) .group(EVENT_LOOP_GROUP) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { if ((connectionSecurityType == SecurityType.TLS)) { ch.pipeline().addLast(clientSslContext.get().newHandler(ch.alloc())); } ch.pipeline().addLast(hf.getDecoders()); ch.pipeline().addLast(hf.getEncoders()); } }); return bootstrap.connect(connection.getDestination()); } private  ConnectFacade(); }
ConnectFacade { public static ChannelFuture createClient(SxpNode node, SxpConnection connection, HandlerFactory hf) { if (!Epoll.isAvailable()) { throw new UnsupportedOperationException(Epoll.unavailabilityCause().getCause()); } SecurityType connectionSecurityType = connection.getSecurityType(); Optional<SslContext> clientSslContext = node.getSslContextFactory().getClientContext(); RecvByteBufAllocator recvByteBufAllocator = new FixedRecvByteBufAllocator(Constants.MESSAGE_LENGTH_MAX); Bootstrap bootstrap = new Bootstrap(); if (connection.hasNonEmptyPassword()) { bootstrap.option(EpollChannelOption.TCP_MD5SIG, Collections.singletonMap(connection.getDestination().getAddress(), connection.getPassword().getBytes(StandardCharsets.US_ASCII))); } else if ((connectionSecurityType == SecurityType.TLS) && !clientSslContext.isPresent()) { throw new IllegalStateException( String.format("%s has TSL enabled but %s does not provide any certificates.", connection, node)); } bootstrap.channel(EpollSocketChannel.class) .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, Configuration.NETTY_CONNECT_TIMEOUT_MILLIS) .option(ChannelOption.RCVBUF_ALLOCATOR, recvByteBufAllocator) .option(ChannelOption.TCP_NODELAY, true) .option(ChannelOption.SO_KEEPALIVE, true) .localAddress(node.getSourceIp().getHostAddress(), 0) .group(EVENT_LOOP_GROUP) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { if ((connectionSecurityType == SecurityType.TLS)) { ch.pipeline().addLast(clientSslContext.get().newHandler(ch.alloc())); } ch.pipeline().addLast(hf.getDecoders()); ch.pipeline().addLast(hf.getEncoders()); } }); return bootstrap.connect(connection.getDestination()); } private  ConnectFacade(); static ChannelFuture createClient(SxpNode node, SxpConnection connection, HandlerFactory hf); static ChannelFuture createServer(SxpNode node, HandlerFactory hf, Map<InetAddress, byte[]> keyMapping); static Map<InetAddress, byte[]> collectAllPasswords(SxpNode node); }
ConnectFacade { public static ChannelFuture createClient(SxpNode node, SxpConnection connection, HandlerFactory hf) { if (!Epoll.isAvailable()) { throw new UnsupportedOperationException(Epoll.unavailabilityCause().getCause()); } SecurityType connectionSecurityType = connection.getSecurityType(); Optional<SslContext> clientSslContext = node.getSslContextFactory().getClientContext(); RecvByteBufAllocator recvByteBufAllocator = new FixedRecvByteBufAllocator(Constants.MESSAGE_LENGTH_MAX); Bootstrap bootstrap = new Bootstrap(); if (connection.hasNonEmptyPassword()) { bootstrap.option(EpollChannelOption.TCP_MD5SIG, Collections.singletonMap(connection.getDestination().getAddress(), connection.getPassword().getBytes(StandardCharsets.US_ASCII))); } else if ((connectionSecurityType == SecurityType.TLS) && !clientSslContext.isPresent()) { throw new IllegalStateException( String.format("%s has TSL enabled but %s does not provide any certificates.", connection, node)); } bootstrap.channel(EpollSocketChannel.class) .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, Configuration.NETTY_CONNECT_TIMEOUT_MILLIS) .option(ChannelOption.RCVBUF_ALLOCATOR, recvByteBufAllocator) .option(ChannelOption.TCP_NODELAY, true) .option(ChannelOption.SO_KEEPALIVE, true) .localAddress(node.getSourceIp().getHostAddress(), 0) .group(EVENT_LOOP_GROUP) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { if ((connectionSecurityType == SecurityType.TLS)) { ch.pipeline().addLast(clientSslContext.get().newHandler(ch.alloc())); } ch.pipeline().addLast(hf.getDecoders()); ch.pipeline().addLast(hf.getEncoders()); } }); return bootstrap.connect(connection.getDestination()); } private  ConnectFacade(); static ChannelFuture createClient(SxpNode node, SxpConnection connection, HandlerFactory hf); static ChannelFuture createServer(SxpNode node, HandlerFactory hf, Map<InetAddress, byte[]> keyMapping); static Map<InetAddress, byte[]> collectAllPasswords(SxpNode node);  }