The content of this file is subject to the terms of the Apache 2.0 open source license + * available at https://www.opensource.org/licenses/apache-2.0 + * + *
Restlet is a registered trademark of QlikTech International AB. + */ +@SuppressWarnings({"requires-automatic", "requires-transitive-automatic"}) +module org.restlet.ext.ziti { + requires transitive org.restlet; + // The OpenZiti SDK jars are not modular and declare no Automatic-Module-Name. + requires transitive ziti; + requires edge.api; + requires io.netty.buffer; + requires io.netty.common; + requires io.netty.transport; + requires io.netty.resolver; + + exports org.restlet.ext.ziti; +} diff --git a/org.restlet.ext.ziti/src/main/java/org/restlet/ext/ziti/ZitiClientTransport.java b/org.restlet.ext.ziti/src/main/java/org/restlet/ext/ziti/ZitiClientTransport.java new file mode 100644 index 0000000000..4b1ebbd02e --- /dev/null +++ b/org.restlet.ext.ziti/src/main/java/org/restlet/ext/ziti/ZitiClientTransport.java @@ -0,0 +1,255 @@ +/** + * Copyright 2005-2026 Qlik + *
+ * The content of this file is subject to the terms of the Apache 2.0 open + * source license available at https://www.opensource.org/licenses/apache-2.0 + *
+ * Restlet is a registered trademark of QlikTech International AB. + */ +package org.restlet.ext.ziti; + +import io.netty.resolver.NoopAddressResolverGroup; +import java.net.InetSocketAddress; +import java.time.Duration; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicLong; +import java.util.logging.Level; +import java.util.logging.Logger; +import org.openziti.ZitiContext; +import org.restlet.Client; +import org.restlet.Context; +import org.restlet.engine.netty.NettyClientTransport; + +/** + * Routes the HTTP client calls to the origins an OpenZiti identity can reach over the Ziti overlay + * network, and the others over TCP. + * + *
An origin is routed over Ziti when the context has a service intercepting its host and port + * ({@link ZitiContext#getService(InetSocketAddress, long)}), such as {@code crm.internal:443}. Its + * host name is then never resolved by the system DNS: the connection is a {@link ZitiSocketChannel} + * that dials that service. Every other origin is reached over TCP. + * + *
The context loads its services after authenticating. Until it has (its status is {@link + * ZitiContext.Status.Active}), routing waits, up to a timeout counted from the creation of the + * transport (10 seconds by default). Past it, the origins are reached over TCP until the services + * are loaded: create the transport early, or wait for the context to be active, so that no + * intercepted host name goes to the DNS. The lookups of the origins are cached, and the cache is + * cleared on each service event of the context. TLS, HTTP/1.1 and HTTP/2 run over it like over TCP: + * an {@code https} origin is encrypted end to end on top of the overlay's own encryption. + * + *
+ * ZitiContext ztx = Ziti.newContext(identityFile, new char[0]); + * Client client = new Client(Protocol.HTTPS); + * ZitiClientTransport.install(client, ztx); + *+ * + *
The context is not owned: destroy it after the client is stopped. The transport listens to the
+ * events of the context: close it once the client is stopped, to stop listening while the context
+ * lives on. Installing a new transport on a client closes the one it replaces.
+ */
+public class ZitiClientTransport implements NettyClientTransport, AutoCloseable {
+
+ private static final Logger LOGGER = Logger.getLogger(ZitiClientTransport.class.getName());
+
+ /** The default time to wait for the services of the context. */
+ private static final Duration DEFAULT_SERVICES_TIMEOUT = Duration.ofSeconds(10);
+
+ /** The maximum number of origins whose lookup is cached. */
+ private static final int MAX_CACHED_ORIGINS = 1024;
+
+ /** The Ziti context. */
+ private final ZitiContext context;
+
+ /** The route over Ziti, shared by every intercepted origin. */
+ private final Route route;
+
+ /** Completed once the context has loaded its services. */
+ private final CompletableFuture
+ * The content of this file is subject to the terms of the Apache 2.0 open
+ * source license available at https://www.opensource.org/licenses/apache-2.0
+ *
+ * Restlet is a registered trademark of QlikTech International AB.
+ */
+package org.restlet.ext.ziti;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.AbstractChannel;
+import io.netty.channel.ChannelConfig;
+import io.netty.channel.ChannelMetadata;
+import io.netty.channel.ChannelOutboundBuffer;
+import io.netty.channel.ChannelPromise;
+import io.netty.channel.ConnectTimeoutException;
+import io.netty.channel.DefaultChannelConfig;
+import io.netty.channel.EventLoop;
+import java.io.IOException;
+import java.net.ConnectException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousSocketChannel;
+import java.nio.channels.ClosedChannelException;
+import java.nio.channels.CompletionHandler;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import org.openziti.ZitiAddress;
+import org.openziti.ZitiContext;
+
+/**
+ * A Netty channel over a connection of the OpenZiti SDK ({@link ZitiContext#open()}), with the
+ * semantics of Netty's own socket channels.
+ *
+ * It replaces {@code org.openziti.netty.ZitiChannel} of ziti-netty 0.35.0, which fires its read
+ * events from the thread completing the read: the SDK releases its pending read before calling the
+ * completion handler, so a read requested during the delivery starts at once, and completes
+ * concurrently on another thread or nested in the call, delivering bytes out of order. Its {@code
+ * config()} also returns a new configuration on each call, so channel options such as {@code
+ * AUTO_READ} are lost, and it blocks the event loop on every write.
+ *
+ * The channel connects to an unresolved {@link InetSocketAddress} such as {@code
+ * crm.internal:443}, or to a {@link ZitiAddress}, both handed to the SDK as is. The SDK dials the
+ * Ziti service intercepting the address, with the destination data and the dial options of its
+ * intercept configuration (such as the identity of the terminator), and fails with an {@link
+ * java.nio.channels.UnresolvedAddressException} if none does. It never resolves a host name with
+ * the system DNS.
+ */
+final class ZitiSocketChannel extends AbstractChannel {
+
+ private static final ChannelMetadata METADATA = new ChannelMetadata(false);
+
+ /** The size of the buffer of a read, the payload size of a Ziti message. */
+ private static final int READ_BUFFER_SIZE = 16 * 1024;
+
+ /** The socket of the Ziti connection. */
+ private final AsynchronousSocketChannel socket;
+
+ private final ChannelConfig config;
+
+ /** Whether the connection is established. */
+ private volatile boolean active;
+
+ /** Whether a read is pending on the socket. Only used on the event loop. */
+ private boolean readPending;
+
+ /** Whether a write is in flight on the socket. Only used on the event loop. */
+ private boolean writeInFlight;
+
+ /**
+ * Constructor.
+ *
+ * @param context The Ziti context dialing the services.
+ */
+ ZitiSocketChannel(ZitiContext context) {
+ super(null);
+ this.socket = context.open();
+ this.config = new DefaultChannelConfig(this);
+ }
+
+ @Override
+ public ChannelConfig config() {
+ return config;
+ }
+
+ @Override
+ public ChannelMetadata metadata() {
+ return METADATA;
+ }
+
+ @Override
+ public boolean isOpen() {
+ return socket.isOpen();
+ }
+
+ @Override
+ public boolean isActive() {
+ return active && socket.isOpen();
+ }
+
+ @Override
+ protected boolean isCompatible(EventLoop loop) {
+ return true;
+ }
+
+ @Override
+ protected SocketAddress localAddress0() {
+ try {
+ return socket.getLocalAddress();
+ } catch (IOException _) {
+ return null;
+ }
+ }
+
+ @Override
+ protected SocketAddress remoteAddress0() {
+ try {
+ return socket.getRemoteAddress();
+ } catch (IOException _) {
+ return null;
+ }
+ }
+
+ @Override
+ protected void doBind(SocketAddress localAddress) {
+ throw new UnsupportedOperationException("A Ziti connection can't be bound");
+ }
+
+ @Override
+ protected void doDisconnect() throws IOException {
+ doClose();
+ }
+
+ @Override
+ protected void doClose() throws IOException {
+ active = false;
+ socket.close();
+ }
+
+ @Override
+ protected Object filterOutboundMessage(Object msg) {
+ if (msg instanceof ByteBuf) {
+ return msg;
+ }
+ throw new UnsupportedOperationException(
+ "Unsupported message type: " + msg.getClass().getName());
+ }
+
+ /** Runs a completion of the SDK on the event loop; returns false if the loop refused it. */
+ private boolean onEventLoop(Runnable task) {
+ try {
+ eventLoop().execute(task);
+ return true;
+ } catch (RejectedExecutionException _) {
+ // The event loop is shutting down, with the channel closed.
+ return false;
+ }
+ }
+
+ @Override
+ protected void doBeginRead() {
+ if (readPending || !isActive()) {
+ return;
+ }
+
+ readPending = true;
+ ByteBuf buffer = config.getAllocator().heapBuffer(READ_BUFFER_SIZE);
+ ByteBuffer target = buffer.nioBuffer(buffer.writerIndex(), buffer.writableBytes());
+
+ try {
+ socket.read(target, null, new ReadHandler(buffer));
+ } catch (RuntimeException e) {
+ readPending = false;
+ buffer.release();
+ throw e;
+ }
+ }
+
+ /** Completes a read on the event loop. */
+ private final class ReadHandler implements CompletionHandler
+ * The content of this file is subject to the terms of the Apache 2.0 open
+ * source license available at https://www.opensource.org/licenses/apache-2.0
+ *
+ * Restlet is a registered trademark of QlikTech International AB.
+ */
+package org.restlet.ext.ziti;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.lang.reflect.Proxy;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.net.SocketOption;
+import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousSocketChannel;
+import java.nio.channels.CompletionHandler;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.openziti.ZitiContext;
+import org.openziti.edge.model.ServiceDetail;
+import org.restlet.Client;
+import org.restlet.Context;
+import org.restlet.Request;
+import org.restlet.Response;
+import org.restlet.Restlet;
+import org.restlet.Server;
+import org.restlet.data.MediaType;
+import org.restlet.data.Method;
+import org.restlet.data.Protocol;
+import org.restlet.data.Status;
+import org.restlet.engine.netty.NettyClientTransport;
+import org.restlet.representation.ByteArrayRepresentation;
+
+/**
+ * Runs the Ziti socket channel under the Restlet client connector. Only the overlay network is
+ * replaced: the fake {@link ZitiContext} intercepts {@code crm.internal}, and its {@link
+ * ZitiContext#open()} returns an asynchronous socket that connects to a local Restlet server
+ * instead of dialing a Ziti service.
+ */
+class ZitiClientTransportTestCase {
+
+ private static final String SERVICE_HOST = "crm.internal";
+
+ private static final int SERVICE_PORT = 8443;
+
+ /** The local port the fake overlay connects the service to. */
+ private int targetPort;
+
+ /** The number of channels opened through the fake overlay. */
+ private final AtomicInteger opened = new AtomicInteger();
+
+ /** Whether the fake context has loaded its services when the transport is created. */
+ private volatile boolean activeAtStart = true;
+
+ /** The number of service lookups made on the fake context. */
+ private final AtomicInteger lookups = new AtomicInteger();
+
+ /** The status listener registered on the fake context. */
+ private volatile Consumer
+ * The content of this file is subject to the terms of the Apache 2.0 open
+ * source license available at https://www.opensource.org/licenses/apache-2.0
+ *
+ * Restlet is a registered trademark of QlikTech International AB.
+ */
+package org.restlet.ext.ziti;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
+import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousServerSocketChannel;
+import java.nio.channels.AsynchronousSocketChannel;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.BooleanSupplier;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+import org.junit.jupiter.api.io.TempDir;
+import org.openziti.Ziti;
+import org.openziti.ZitiAddress;
+import org.openziti.ZitiContext;
+import org.openziti.edge.model.ServiceDetail;
+import org.restlet.Client;
+import org.restlet.Context;
+import org.restlet.Request;
+import org.restlet.Response;
+import org.restlet.Restlet;
+import org.restlet.Server;
+import org.restlet.data.MediaType;
+import org.restlet.data.Method;
+import org.restlet.data.Protocol;
+import org.restlet.data.Status;
+import org.restlet.representation.ByteArrayRepresentation;
+
+/**
+ * Calls a Restlet server over a real OpenZiti network with {@link ZitiClientTransport}.
+ *
+ * Runs only when given two enrolled identities: {@code -Dziti.client.identity=client.json},
+ * which may dial the {@code crm} service, intercepted as {@code crm.internal:8443}, and {@code
+ * -Dziti.server.identity=server.json}, which may bind it. The server identity binds the service
+ * with the SDK's server API and pipes each Ziti connection to a local Restlet server, so a request
+ * goes: Restlet client, Ziti socket channel, edge router, Ziti server socket, Restlet server.
+ */
+@EnabledIfSystemProperty(named = "ziti.client.identity", matches = ".+")
+class ZitiNetworkTestCase {
+
+ private static final String SERVICE = "crm";
+
+ private static final String HOST = "crm.internal";
+
+ private static ZitiContext clientZiti;
+
+ private static ZitiContext serverZiti;
+
+ private static AsynchronousServerSocketChannel binding;
+
+ private static ExecutorService bridge;
+
+ /** The local port the bridge pipes the Ziti connections to. */
+ private static volatile int targetPort;
+
+ private static Server server;
+
+ @BeforeAll
+ static void start() throws Exception {
+ clientZiti =
+ Ziti.newContext(new File(System.getProperty("ziti.client.identity")), new char[0]);
+ serverZiti =
+ Ziti.newContext(new File(System.getProperty("ziti.server.identity")), new char[0]);
+
+ // A context learns its services asynchronously after authenticating: binding before the
+ // server identity has loaded the service is refused as "no permission to bind".
+ awaitUntil(
+ () -> loaded(() -> serverZiti.getService(SERVICE, 0L)) != null,
+ "The server never saw the crm service");
+
+ // Host the service: accept the Ziti connections and pipe them to the local server.
+ binding = serverZiti.openServer();
+ binding.bind(new ZitiAddress.Bind(SERVICE, null, false), 64);
+ bridge = Executors.newVirtualThreadPerTaskExecutor();
+ bridge.submit(ZitiNetworkTestCase::acceptLoop);
+
+ // The client learns its services asynchronously after authenticating, and the bind
+ // registers a terminator asynchronously: dialing before gives "no terminators".
+ awaitReachable(InetSocketAddress.createUnresolved(HOST, 8443));
+ }
+
+ /**
+ * Waits until the client sees a service intercepting an address, with a terminator.
+ *
+ * @param address The intercepted address.
+ */
+ private static void awaitReachable(InetSocketAddress address) throws Exception {
+ awaitUntil(
+ () -> {
+ ServiceDetail service = loaded(() -> clientZiti.getService(address, 0L));
+ return service != null && !clientZiti.getServiceTerminators(service).isEmpty();
+ },
+ "The service of " + address + " is not reachable");
+ }
+
+ /**
+ * Looks a service up among the loaded ones, without waiting: the overloads without timeout wait
+ * for a service event, forever if the identity has none.
+ *
+ * @param lookup A lookup with a zero timeout.
+ * @return The service, or null if not loaded.
+ */
+ private static ServiceDetail loaded(ServiceLookup lookup) {
+ try {
+ return lookup.get();
+ } catch (TimeoutException _) {
+ return null;
+ }
+ }
+
+ /** A service lookup declaring the TimeoutException that the SDK throws undeclared. */
+ @FunctionalInterface
+ private interface ServiceLookup {
+ ServiceDetail get() throws TimeoutException;
+ }
+
+ /**
+ * Waits up to a minute for a condition on the network, checked every 250 ms.
+ *
+ * @param condition The condition.
+ * @param message The failure message.
+ */
+ private static void awaitUntil(BooleanSupplier condition, String message) throws Exception {
+ CompletableFuture
+ * The content of this file is subject to the terms of the Apache 2.0 open
+ * source license available at https://www.opensource.org/licenses/apache-2.0
+ *
+ * Restlet is a registered trademark of QlikTech International AB.
+ */
+package org.restlet.ext.ziti;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.ConnectTimeoutException;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.MultiThreadIoEventLoopGroup;
+import io.netty.channel.nio.NioIoHandler;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.lang.reflect.Proxy;
+import java.net.ConnectException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.net.SocketOption;
+import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousSocketChannel;
+import java.nio.channels.CompletionHandler;
+import java.nio.channels.UnresolvedAddressException;
+import java.nio.channels.spi.AsynchronousChannelProvider;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayDeque;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.openziti.ZitiAddress;
+import org.openziti.ZitiContext;
+import org.openziti.edge.model.ServiceDetail;
+
+/**
+ * Checks that {@link ZitiSocketChannel} gives a Ziti connection the semantics of Netty's own socket
+ * channels, over a scripted SDK socket.
+ */
+class ZitiSocketChannelTestCase {
+
+ private static final InetSocketAddress SERVICE =
+ InetSocketAddress.createUnresolved("crm.internal", 443);
+
+ private EventLoopGroup group;
+
+ private ScriptedSocket socket;
+
+ private ZitiSocketChannel channel;
+
+ /** What the pipeline received. */
+ private final Recorder recorder = new Recorder();
+
+ @BeforeEach
+ void start() throws Exception {
+ group = new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory());
+ socket = new ScriptedSocket();
+ channel = new ZitiSocketChannel(fakeZiti(socket));
+ channel.config().setAutoRead(false);
+ channel.pipeline().addLast(recorder);
+ group.register(channel).sync();
+ }
+
+ @AfterEach
+ void stop() throws Exception {
+ channel.close().sync();
+ group.shutdownGracefully(0, 1, TimeUnit.SECONDS).sync();
+ }
+
+ /** A Ziti context intercepting crm.internal:443 with the "crm" service. */
+ private static ZitiContext fakeZiti(AsynchronousSocketChannel socket) {
+ return (ZitiContext)
+ Proxy.newProxyInstance(
+ ZitiContext.class.getClassLoader(),
+ new Class[] {ZitiContext.class},
+ (proxy, method, args) ->
+ switch (method.getName()) {
+ case "getService" ->
+ SERVICE.equals(args[0])
+ ? new ServiceDetail().name("crm")
+ : null;
+ case "open" -> socket;
+ case "toString" -> "fake-ziti";
+ case "hashCode" -> System.identityHashCode(proxy);
+ case "equals" -> proxy == args[0];
+ default ->
+ throw new UnsupportedOperationException(
+ method.getName());
+ });
+ }
+
+ /** Runs a task on the event loop and waits for it. */
+ private void onEventLoop(Runnable task) throws Exception {
+ channel.eventLoop().submit(task).get(5, TimeUnit.SECONDS);
+ }
+
+ private void connect() throws Exception {
+ channel.connect(SERVICE).sync();
+ }
+
+ @Test
+ void letsTheSdkDialTheServiceInterceptingTheOrigin() throws Exception {
+ connect();
+
+ assertTrue(channel.isActive());
+ // The SDK maps it to the service, with the dial data and options of its intercept.
+ assertSame(SERVICE, socket.dialed);
+ assertTrue(recorder.active.get());
+ }
+
+ @Test
+ void dialsAZitiAddressAsIs() throws Exception {
+ ZitiAddress.Dial address = new ZitiAddress.Dial("billing", null, null, null);
+
+ channel.connect(address).sync();
+
+ assertSame(address, socket.dialed);
+ }
+
+ @Test
+ void failsToConnectToAnOriginWithoutService() {
+ ChannelFuture future =
+ channel.connect(InetSocketAddress.createUnresolved("other.internal", 443))
+ .awaitUninterruptibly();
+
+ assertInstanceOf(UnresolvedAddressException.class, future.cause());
+ assertFalse(channel.isActive());
+ }
+
+ @Test
+ void reportsAConnectFailure() {
+ socket.connectFailure = new IOException("no terminators");
+
+ ChannelFuture future = channel.connect(SERVICE).awaitUninterruptibly();
+
+ ConnectException failure = assertInstanceOf(ConnectException.class, future.cause());
+ assertSame(socket.connectFailure, failure.getCause());
+ assertFalse(channel.isActive());
+ }
+
+ @Test
+ void timesOutAConnect() {
+ socket.connectHangs = true;
+ channel.config().setOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, 100);
+
+ ChannelFuture future = channel.connect(SERVICE).awaitUninterruptibly();
+
+ assertInstanceOf(ConnectTimeoutException.class, future.cause());
+ assertTrue(channel.closeFuture().awaitUninterruptibly(5, TimeUnit.SECONDS));
+ }
+
+ @Test
+ void refusesToBind() {
+ ChannelFuture future = channel.bind(new InetSocketAddress(0)).awaitUninterruptibly();
+
+ assertInstanceOf(UnsupportedOperationException.class, future.cause());
+ }
+
+ @Test
+ void readsOneAtATime() throws Exception {
+ connect();
+
+ onEventLoop(
+ () -> {
+ channel.read();
+ channel.read();
+ channel.read();
+ });
+
+ assertEquals(1, socket.reads.get(), "A read requested while one is pending is ignored");
+
+ // Completed by an SDK thread: the bytes are fired on the event loop.
+ socket.completePendingRead("hello");
+ recorder.received.get(5, TimeUnit.SECONDS);
+
+ assertEquals("hello", recorder.text());
+ assertTrue(recorder.allOnEventLoop.get());
+
+ onEventLoop(channel::read);
+ assertEquals(2, socket.reads.get(), "A read is accepted once the pending one completed");
+ }
+
+ @Test
+ void neverFiresAReadNestedInTheReadRequest() throws Exception {
+ connect();
+ // The SDK completes the read inside read(), the data being buffered already.
+ socket.bufferedReads.add("buffered");
+ AtomicBoolean inRead = new AtomicBoolean();
+ recorder.onRead = bytes -> recorder.nested.compareAndSet(false, inRead.get());
+
+ onEventLoop(
+ () -> {
+ inRead.set(true);
+ channel.read();
+ inRead.set(false);
+ });
+ recorder.received.get(5, TimeUnit.SECONDS);
+
+ assertEquals("buffered", recorder.text());
+ assertFalse(recorder.nested.get(), "The read was fired nested in Channel.read()");
+ }
+
+ @Test
+ void readsOnItsOwnWithAutoRead() throws Exception {
+ connect();
+ socket.bufferedReads.add("first");
+ socket.bufferedReads.add("second");
+
+ channel.config().setAutoRead(true);
+ recorder.awaitText("firstsecond");
+
+ assertEquals(3, socket.reads.get(), "A read is started after each one");
+ }
+
+ @Test
+ void closesAtTheEndOfTheStream() throws Exception {
+ connect();
+ socket.bufferedReads.add(-1);
+
+ onEventLoop(channel::read);
+
+ assertTrue(channel.closeFuture().await(5, TimeUnit.SECONDS));
+ // Netty fires channelInactive after completing the close future.
+ assertTrue(recorder.inactive.await(5, TimeUnit.SECONDS));
+ assertFalse(recorder.active.get());
+ }
+
+ @Test
+ void reportsAReadFailure() throws Exception {
+ connect();
+ IOException failure = new IOException("connection reset");
+ socket.bufferedReads.add(failure);
+
+ onEventLoop(channel::read);
+
+ assertTrue(channel.closeFuture().await(5, TimeUnit.SECONDS));
+ assertSame(failure, recorder.failures.get(0));
+ }
+
+ @Test
+ void writesInOrderWithoutBlocking() throws Exception {
+ connect();
+ // The SDK accepts 3 bytes per write, completing the write inside the call.
+ socket.maxWrite = 3;
+
+ channel.write(Unpooled.copiedBuffer("hello ", StandardCharsets.UTF_8));
+ channel.write(Unpooled.EMPTY_BUFFER);
+ ChannelFuture last =
+ channel.writeAndFlush(Unpooled.copiedBuffer("over ziti", StandardCharsets.UTF_8));
+
+ assertTrue(last.await(5, TimeUnit.SECONDS));
+ assertTrue(last.isSuccess());
+ assertArrayEquals(
+ "hello over ziti".getBytes(StandardCharsets.UTF_8), socket.written.toByteArray());
+ assertEquals(1, socket.maxWritesInFlight.get(), "One write is in flight at a time");
+ }
+
+ @Test
+ void resumesOnlyTheFlushedWrites() throws Exception {
+ connect();
+ socket.holdWrites = true;
+
+ ChannelFuture first =
+ channel.writeAndFlush(Unpooled.copiedBuffer("first", StandardCharsets.UTF_8));
+ ChannelFuture second =
+ channel.write(Unpooled.copiedBuffer("second", StandardCharsets.UTF_8));
+ // Both writes reached the event loop: the first is in flight, the second queued.
+ onEventLoop(() -> {});
+ socket.holdWrites = false;
+ socket.completeHeldWrite();
+
+ assertTrue(first.await(5, TimeUnit.SECONDS));
+ onEventLoop(() -> {});
+ assertFalse(second.isDone(), "A write not flushed was sent");
+ assertEquals("first", socket.writtenText());
+
+ channel.flush();
+ assertTrue(second.await(5, TimeUnit.SECONDS));
+ assertEquals("firstsecond", socket.writtenText());
+ }
+
+ @Test
+ void reportsAWriteFailure() throws Exception {
+ connect();
+ socket.writeFailure = new IOException("broken pipe");
+
+ ChannelFuture future =
+ channel.writeAndFlush(Unpooled.copiedBuffer("lost", StandardCharsets.UTF_8))
+ .awaitUninterruptibly();
+
+ assertFalse(future.isSuccess());
+ assertTrue(channel.closeFuture().await(5, TimeUnit.SECONDS));
+ assertSame(socket.writeFailure, recorder.failures.get(0));
+ }
+
+ @Test
+ void rejectsMessagesOtherThanBuffers() throws Exception {
+ connect();
+
+ ChannelFuture future = channel.writeAndFlush("text").awaitUninterruptibly();
+
+ assertInstanceOf(UnsupportedOperationException.class, future.cause());
+ }
+
+ /** Records what reaches the end of the pipeline. */
+ private final class Recorder extends ChannelInboundHandlerAdapter {
+
+ private final ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+
+ final CompletableFuture
+ *
+ *
+ *