### Question:
TcpConnector { public static <C extends ListenableAsyncCloseable> Single<C> connect( final @Nullable SocketAddress localAddress, final Object resolvedRemoteAddress, final ReadOnlyTcpClientConfig config, final boolean autoRead, final ExecutionContext executionContext, final Function<Channel, Single<? extends C>> connectionFactory) { requireNonNull(resolvedRemoteAddress); requireNonNull(config); requireNonNull(executionContext); return new SubscribableSingle<C>() { @Override protected void handleSubscribe(final Subscriber<? super C> subscriber) { ConnectHandler<C> connectHandler = new ConnectHandler<>(subscriber, connectionFactory); try { Future<?> connectFuture = connect0(localAddress, resolvedRemoteAddress, config, autoRead, executionContext, connectHandler); connectHandler.connectFuture(connectFuture); connectFuture.addListener(f -> { Throwable cause = f.cause(); if (cause != null) { if (cause instanceof ConnectTimeoutException) { String msg = resolvedRemoteAddress instanceof FileDescriptorSocketAddress ? "Failed to register: " + resolvedRemoteAddress : "Failed to connect: " + resolvedRemoteAddress + " (localAddress: " + localAddress + ")"; cause = new io.servicetalk.client.api.ConnectTimeoutException(msg, cause); } else if (cause instanceof ConnectException) { cause = new RetryableConnectException((ConnectException) cause); } connectHandler.connectFailed(cause); } }); } catch (Throwable t) { connectHandler.unexpectedFailure(t); } } }; } private  TcpConnector(); static Single<C> connect(
            final @Nullable SocketAddress localAddress, final Object resolvedRemoteAddress,
            final ReadOnlyTcpClientConfig config, final boolean autoRead, final ExecutionContext executionContext,
            final Function<Channel, Single<? extends C>> connectionFactory);  }### Answer:
@Test public void testConnect() throws Exception { client.connectBlocking(CLIENT_CTX, serverAddress); }
@Test public void testRegisteredAndActiveEventsFired() throws Exception { final CountDownLatch registeredLatch = new CountDownLatch(1); final CountDownLatch activeLatch = new CountDownLatch(1); NettyConnection<Buffer, Buffer> connection = TcpConnector.<NettyConnection<Buffer, Buffer>>connect(null, serverContext.listenAddress(), new TcpClientConfig().asReadOnly(emptyList()), false, CLIENT_CTX, channel -> DefaultNettyConnection.initChannel(channel, CLIENT_CTX.bufferAllocator(), CLIENT_CTX.executor(), o -> true, UNSUPPORTED_PROTOCOL_CLOSE_HANDLER, defaultFlushStrategy(), null, channel2 -> { channel2.pipeline().addLast(new ChannelInboundHandlerAdapter() { @Override public void channelRegistered(ChannelHandlerContext ctx) { registeredLatch.countDown(); ctx.fireChannelRegistered(); } @Override public void channelActive(ChannelHandlerContext ctx) { activeLatch.countDown(); ctx.fireChannelActive(); } }); }, CLIENT_CTX.executionStrategy(), mock(Protocol.class), NoopConnectionObserver.INSTANCE)) .toFuture().get(); connection.closeAsync().toFuture().get(); registeredLatch.await(); activeLatch.await(); }