Skip to content

Complete client disconnect when the channel closes #2135

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kevinherron merged 2 commits into from
Oct 23, 2026
Merged

Complete client disconnect when the channel closes #2135

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8093d88
Complete client disconnect when the channel closes
kevinherron Oct 23, 2026
8551915
Update netty-channel-fsm to 1.0.4
kevinherron Oct 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 13 additions & 23 deletions ...in/java/org/eclipse/milo/opcua/stack/transport/websocket/OpcWebSocketClientTransport.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -256,27 +252,21 @@ protected void initChannel(SocketChannel channel) throws Exception {
public CompletableFuture<Void> disconnect(FsmContext<State, Event> ctx, Channel channel) {
var disconnectFuture = new CompletableFuture<Void>();

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 =
Expand Down
38 changes: 14 additions & 24 deletions ...rc/main/java/org/eclipse/milo/opcua/stack/transport/client/tcp/OpcTcpClientTransport.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -365,29 +361,23 @@ protected void initChannel(SocketChannel ch) {
public CompletableFuture<Void> disconnect(FsmContext<State, Event> ctx, Channel channel) {
var disconnectFuture = new CompletableFuture<Void>();

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 =
Expand Down
105 changes: 105 additions & 0 deletions .../src/test/java/org/eclipse/milo/opcua/stack/transport/client/tcp/OpcTcpTransportTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,13 +14,18 @@
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;
import static org.junit.jupiter.api.Assertions.fail;

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;
Expand All @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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<Unit> 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.
Expand Down
2 changes: 1 addition & 1 deletion pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,7 @@
<jakarta-xml-bind-api.version>4.0.5</jakarta-xml-bind-api.version>
<jaxb.version>4.0.9</jaxb.version>
<netty.version>4.1.139.Final</netty.version>
<netty-channel-fsm.version>1.0.2</netty-channel-fsm.version>
<netty-channel-fsm.version>1.0.4</netty-channel-fsm.version>
<slf4j.version>2.0.20</slf4j.version>
<jspecify.version>1.0.1</jspecify.version>
<xmlunit.version>2.14.0</xmlunit.version>
Expand Down
Loading