diff --git a/opc-ua-stack/transport-websocket/src/main/java/org/eclipse/milo/opcua/stack/transport/websocket/OpcWebSocketClientTransport.java b/opc-ua-stack/transport-websocket/src/main/java/org/eclipse/milo/opcua/stack/transport/websocket/OpcWebSocketClientTransport.java index 5e9ff4ad11..5eca42b843 100644 --- a/opc-ua-stack/transport-websocket/src/main/java/org/eclipse/milo/opcua/stack/transport/websocket/OpcWebSocketClientTransport.java +++ b/opc-ua-stack/transport-websocket/src/main/java/org/eclipse/milo/opcua/stack/transport/websocket/OpcWebSocketClientTransport.java @@ -24,9 +24,6 @@ import io.netty.buffer.PooledByteBufAllocator; import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; -import io.netty.channel.ChannelFutureListener; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.ConnectTimeoutException; @@ -45,7 +42,6 @@ import io.netty.handler.ssl.SslContextBuilder; import io.netty.handler.ssl.util.InsecureTrustManagerFactory; import io.netty.util.Timeout; -import io.netty.util.TimerTask; import java.net.ConnectException; import java.net.InetSocketAddress; import java.net.URI; @@ -256,27 +252,21 @@ protected void initChannel(SocketChannel channel) throws Exception { public CompletableFuture disconnect(FsmContext ctx, Channel channel) { var disconnectFuture = new CompletableFuture(); - TimerTask onTimeout = - t -> - channel - .close() - .addListener( - (ChannelFutureListener) channelFuture -> disconnectFuture.complete(null)); - - Timeout timeout = config.getWheelTimer().newTimeout(onTimeout, 5, TimeUnit.SECONDS); + // Close the channel ourselves if the server doesn't close it after CloseSecureChannel. + Timeout timeout = + config.getWheelTimer().newTimeout(t -> channel.close(), 5, TimeUnit.SECONDS); + // Netty notifies a close future listener even when it's added after the channel closed, so + // the disconnect completes without waiting for the timeout when this action runs against a + // channel that is already closed or closing. channel - .pipeline() - .addFirst( - new ChannelInboundHandlerAdapter() { - @Override - public void channelInactive(ChannelHandlerContext channelContext) throws Exception { - // TODO MDC? - logger.debug("channelInactive() disconnect complete"); - timeout.cancel(); - disconnectFuture.complete(null); - super.channelInactive(channelContext); - } + .closeFuture() + .addListener( + f -> { + // TODO MDC? + logger.debug("Channel closed, disconnect complete"); + timeout.cancel(); + disconnectFuture.complete(null); }); var requestHeader = diff --git a/opc-ua-stack/transport/src/main/java/org/eclipse/milo/opcua/stack/transport/client/tcp/OpcTcpClientTransport.java b/opc-ua-stack/transport/src/main/java/org/eclipse/milo/opcua/stack/transport/client/tcp/OpcTcpClientTransport.java index e1805cc491..6b5951c73f 100644 --- a/opc-ua-stack/transport/src/main/java/org/eclipse/milo/opcua/stack/transport/client/tcp/OpcTcpClientTransport.java +++ b/opc-ua-stack/transport/src/main/java/org/eclipse/milo/opcua/stack/transport/client/tcp/OpcTcpClientTransport.java @@ -23,16 +23,12 @@ import io.netty.buffer.PooledByteBufAllocator; import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; -import io.netty.channel.ChannelFutureListener; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.ConnectTimeoutException; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.util.Timeout; -import io.netty.util.TimerTask; import java.net.ConnectException; import java.net.InetSocketAddress; import java.util.List; @@ -365,29 +361,23 @@ protected void initChannel(SocketChannel ch) { public CompletableFuture disconnect(FsmContext ctx, Channel channel) { var disconnectFuture = new CompletableFuture(); - TimerTask onTimeout = - t -> - channel - .close() - .addListener( - (ChannelFutureListener) channelFuture -> disconnectFuture.complete(null)); - - Timeout timeout = config.getWheelTimer().newTimeout(onTimeout, 5, TimeUnit.SECONDS); + // Close the channel ourselves if the server doesn't close it after CloseSecureChannel. + Timeout timeout = + config.getWheelTimer().newTimeout(t -> channel.close(), 5, TimeUnit.SECONDS); + // Netty notifies a close future listener even when it's added after the channel closed, so + // the disconnect completes without waiting for the timeout when this action runs against a + // channel that is already closed or closing. channel - .pipeline() - .addFirst( - new ChannelInboundHandlerAdapter() { - @Override - public void channelInactive(ChannelHandlerContext channelContext) throws Exception { - try (MDC.MDCCloseable ignored = MDC.putCloseable("instance-id", instanceId)) { - logger.debug("channelInactive() disconnect complete"); - } - - timeout.cancel(); - disconnectFuture.complete(null); - super.channelInactive(channelContext); + .closeFuture() + .addListener( + f -> { + try (MDC.MDCCloseable ignored = MDC.putCloseable("instance-id", instanceId)) { + logger.debug("Channel closed, disconnect complete"); } + + timeout.cancel(); + disconnectFuture.complete(null); }); var requestHeader = diff --git a/opc-ua-stack/transport/src/test/java/org/eclipse/milo/opcua/stack/transport/client/tcp/OpcTcpTransportTest.java b/opc-ua-stack/transport/src/test/java/org/eclipse/milo/opcua/stack/transport/client/tcp/OpcTcpTransportTest.java index 588102164d..3d674031db 100644 --- a/opc-ua-stack/transport/src/test/java/org/eclipse/milo/opcua/stack/transport/client/tcp/OpcTcpTransportTest.java +++ b/opc-ua-stack/transport/src/test/java/org/eclipse/milo/opcua/stack/transport/client/tcp/OpcTcpTransportTest.java @@ -14,6 +14,7 @@ import static org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint; import static org.eclipse.milo.opcua.stack.transport.TestPortAllocator.allocatePort; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -21,6 +22,10 @@ import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; +import io.netty.channel.Channel; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelOutboundHandlerAdapter; +import io.netty.channel.ChannelPromise; import java.io.EOFException; import java.io.IOException; import java.io.InputStream; @@ -38,6 +43,8 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.stream.Stream; @@ -89,6 +96,7 @@ import org.eclipse.milo.opcua.stack.core.types.structured.UserTokenPolicy; import org.eclipse.milo.opcua.stack.core.util.SelfSignedCertificateBuilder; import org.eclipse.milo.opcua.stack.core.util.SelfSignedCertificateGenerator; +import org.eclipse.milo.opcua.stack.core.util.Unit; import org.eclipse.milo.opcua.stack.transport.client.ClientApplicationContext; import org.eclipse.milo.opcua.stack.transport.client.uasc.ClientSecureChannel; import org.eclipse.milo.opcua.stack.transport.server.OpcServerTransport; @@ -416,6 +424,103 @@ public UInteger getRequestTimeout() { LOGGER.debug("server transport unbound"); } + // The disconnect action falls back to closing the channel after 5 seconds, so these tests bound + // disconnect() well inside that timeout to show the channel close completed it. + @Test + void disconnectSendsCloseSecureChannelAndCompletesWhenChannelCloses() throws Exception { + OpcServerTransport serverTransport = + bindServerTransport(SecurityPolicy.None, MessageSecurityMode.None); + + var closeSecureChannelSent = new CountDownLatch(1); + + OpcTcpClientTransportConfig config = + OpcTcpClientTransportConfig.newBuilder() + .setChannelPipelineCustomizer( + pipeline -> + pipeline.addFirst( + new ChannelOutboundHandlerAdapter() { + @Override + public void write( + ChannelHandlerContext ctx, Object msg, ChannelPromise promise) + throws Exception { + + if (msg instanceof ByteBuf buffer + && buffer.readableBytes() >= 3 + && buffer.getByte(buffer.readerIndex()) == 'C' + && buffer.getByte(buffer.readerIndex() + 1) == 'L' + && buffer.getByte(buffer.readerIndex() + 2) == 'O') { + + closeSecureChannelSent.countDown(); + } + super.write(ctx, msg, promise); + } + })) + .build(); + + var transport = new OpcTcpClientTransport(config); + + try { + transport.connect(noSecurityClientApplicationContext()).get(); + Channel channel = transport.getChannelFsm().getChannel().get(); + + transport.disconnect().get(2, TimeUnit.SECONDS); + + assertEquals(0, closeSecureChannelSent.getCount()); + assertFalse(channel.isOpen()); + } finally { + serverTransport.unbind(); + } + } + + // When the application calls disconnect() at about the same moment the connection closes, the + // ChannelFsm can evaluate Disconnect before the ChannelInactive event queued by the close. The + // disconnect action then runs against a channel that is already closed. + @Test + void disconnectCompletesWhenChannelClosedBeforeDisconnectActionRuns() throws Exception { + OpcServerTransport serverTransport = + bindServerTransport(SecurityPolicy.None, MessageSecurityMode.None); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + + try { + var transport = + new OpcTcpClientTransport( + OpcTcpClientTransportConfig.newBuilder().setExecutor(executor).build()); + + transport.connect(noSecurityClientApplicationContext()).get(); + Channel channel = transport.getChannelFsm().getChannel().get(); + + // Hold the ChannelFsm's only executor thread so Disconnect is queued before the + // ChannelInactive event from the close below. + var release = new CountDownLatch(1); + executor.execute( + () -> { + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + + CompletableFuture disconnected = transport.disconnect(); + channel.close().sync(); + release.countDown(); + + disconnected.get(2, TimeUnit.SECONDS); + } finally { + executor.shutdownNow(); + serverTransport.unbind(); + } + } + + private ClientApplicationContext noSecurityClientApplicationContext() { + return clientApplicationContext( + SecurityPolicy.None, + new CertificateMaterial(clientKeyPair, clientCertificate, clientCertificateBytes), + new CertificateMaterial(serverKeyPair, serverCertificate, serverCertificateBytes), + MessageSecurityMode.None); + } + @Test void securityPolicyRejectedOnUnsecuredChannel() throws Exception { // We opened an unsecured channel even though only secured endpoints are available. diff --git a/pom.xml b/pom.xml index fef4eab1ce..b50a770da8 100644 --- a/pom.xml +++ b/pom.xml @@ -66,7 +66,7 @@ 4.0.5 4.0.9 4.1.139.Final - 1.0.2 + 1.0.4 2.0.20 1.0.1 2.14.0