Skip to content

Close websocket sessions reliably when the upstream is missing #176

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
vharseko merged 4 commits into from
Oct 23, 2026
Merged

Close websocket sessions reliably when the upstream is missing #176

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
f0fdcf7
Close websocket sessions reliably when the upstream is missing
vharseko Sep 18, 2026
1a122ed
Add @Override to the upstream message handlers in ServerEndPoint
vharseko Sep 18, 2026
f3c8770
Address review on websocket close handling
vharseko Oct 22, 2026
d1a4ade
Address second review round on websocket close handling
vharseko Oct 22, 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
12 changes: 5 additions & 7 deletions openig-core/src/main/java/org/forgerock/openig/text/SeparatedValuesFile.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
*
* Copyright 2010-2011 ApexIdentity Inc.
* Portions Copyright 2011-2015 ForgeRock AS.
* Portions Copyright 2026 3A Systems, LLC.
*/

package org.forgerock.openig.text;
Expand Down Expand Up @@ -148,11 +149,10 @@ public List<String> getFields() {
*/
public Map<String, String> getRecord(String key, String value) throws IOException {
Map<String, String> map = null;
SeparatedValuesReader reader = new SeparatedValuesReader(
new InputStreamReader(new FileInputStream(file), charset),
separator
);
try {
try (FileInputStream stream = new FileInputStream(file);
SeparatedValuesReader reader = new SeparatedValuesReader(
new InputStreamReader(stream, charset),
separator)) {
List<String> fields = this.fields;
if (header) {
// first line in the file is the field header
Expand Down Expand Up @@ -181,8 +181,6 @@ public Map<String, String> getRecord(String key, String value) throws IOExceptio
}
}
}
} finally {
reader.close();
}
return map;
}
Expand Down
5 changes: 4 additions & 1 deletion openig-core/src/main/java/org/forgerock/openig/text/SeparatedValuesReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,12 +13,14 @@
*
* Copyright 2010-2011 ApexIdentity Inc.
* Portions Copyright 2011-2016 ForgeRock AS.
* Portions Copyright 2026 3A Systems, LLC.
*/

package org.forgerock.openig.text;

import static org.forgerock.util.Utils.closeSilently;

import java.io.Closeable;
import java.io.IOException;
import java.io.Reader;
import java.util.ArrayList;
Expand All @@ -27,7 +29,7 @@
/**
* Reads records with delimiter-separated values from a character stream.
*/
public class SeparatedValuesReader {
public class SeparatedValuesReader implements Closeable {

private static final int CR = '\r';
private static final int LF = '\n';
Expand Down Expand Up @@ -113,6 +115,7 @@ public List<String> next() throws IOException {
* reader has been closed, further {@code next()} invocations will throw an
* {@code IOException}. Closing a previously closed reader has no effect.
*/
@Override
public void close() {
closeSilently(input);
}
Expand Down
128 changes: 81 additions & 47 deletions openig-core/src/main/java/org/openidentityplatform/openig/websocket/ServerEndPoint.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
* Header, with the fields enclosed by brackets [] replaced by your own identifying
* information: "Portions copyright [year] [name of copyright owner]".
*
* Copyright 2022-2025 3A Systems LLC.
* Copyright 2022-2026 3A Systems LLC.
*/

package org.openidentityplatform.openig.websocket;
Expand All @@ -35,6 +35,7 @@
import jakarta.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
Expand All @@ -45,7 +46,67 @@ public class ServerEndPoint {
static Logger logger=LoggerFactory.getLogger(ServerEndPoint.class);

Principal principal=null;
Session session_upstream=null;
volatile Session session_upstream=null;

/** A close frame reason phrase is limited to 123 UTF-8 bytes; {@link CloseReason} rejects longer ones. */
static final int MAX_REASON_PHRASE_BYTES = 123;

/**
* Builds a {@link CloseReason}, truncating the phrase so that it never exceeds the protocol limit
* and never ends in the first half of a surrogate pair. A truncated phrase ends with U+2026.
*/
static CloseReason closeReason(CloseReason.CloseCode code, String phrase) {
if (phrase.getBytes(StandardCharsets.UTF_8).length <= MAX_REASON_PHRASE_BYTES) {
return new CloseReason(code, truncate(phrase, MAX_REASON_PHRASE_BYTES));
}
// mark the cut as Tomcat does: at most 120 bytes, then U+2026 (3 bytes in UTF-8)
return new CloseReason(code, truncate(phrase, MAX_REASON_PHRASE_BYTES - 3) + "\u2026");
}

/** Cuts the phrase to at most {@code maxBytes} UTF-8 bytes without splitting a character. */
private static String truncate(String phrase, int maxBytes) {
String reason = phrase.length() > maxBytes ? phrase.substring(0, maxBytes) : phrase;
// a lone high surrogate is encoded as a single '?' byte, so the length check alone does not catch it
while (reason.getBytes(StandardCharsets.UTF_8).length > maxBytes
|| (!reason.isEmpty() && Character.isHighSurrogate(reason.charAt(reason.length() - 1)))) {
reason = reason.substring(0, reason.length() - 1);
}
return reason;
}

/** Waits (up to ~5 s) for the upstream session to be connected and returns it. */
private Session upstream() throws IOException, InterruptedException {
int ct=0;
Session upstream=session_upstream;
while ((ct++<5000)&&(upstream==null || !upstream.isOpen())) {
Thread.sleep(1);
upstream=session_upstream;
}
if (upstream==null || !upstream.isOpen()) {
throw new IOException("upstream not connected");
}
return upstream;
}

/** Closes the upstream session (if any) with the given reason and forgets it. */
private void closeUpstream(CloseReason reason) {
final Session upstream=session_upstream;
session_upstream=null;
if (upstream!=null) {
try {
upstream.close(reason);
} catch (Throwable e1) {}
}
}

/** Closes both the client and the upstream sessions abnormally with the given reason. */
private void closeAbnormally(Session session_client, String phrase) {
final CloseReason reason=closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY, phrase);
try {
session_client.close(reason);
} catch (Throwable e1) {}
closeUpstream(reason);
}

public void authorize() throws Exception {
if (principal==null || principal.authorize().getCode()!=101) {
Expand Down Expand Up @@ -93,6 +154,7 @@ public void onOpen(Session session_upstream, EndpointConfig config) {
session_upstream.setMaxTextMessageBufferSize(session_client.getMaxTextMessageBufferSize());
session_upstream.setMaxIdleTimeout(session_client.getMaxIdleTimeout());
session_upstream.addMessageHandler(new MessageHandler.Whole<String>() {
@Override
public void onMessage(String message) {
try {
if (logger.isTraceEnabled()) {
Expand All @@ -101,18 +163,20 @@ public void onMessage(String message) {
authorize();
session_client.getBasicRemote().sendText(message);
} catch (Throwable e) {
final CloseReason reason=closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"upstream message error: "+e.toString());
try {
session_client.close(new CloseReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"upstream message error: "+e.toString()));
session_client.close(reason);
} catch (Throwable e1) {}
try {
session_upstream.close(new CloseReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"upstream message error: "+e.toString()));
session_upstream.close(reason);
} catch (Throwable e1) {}

}
}
});

session_upstream.addMessageHandler(new MessageHandler.Whole<ByteBuffer>() {
@Override
public void onMessage(ByteBuffer message) {
try {
if (logger.isTraceEnabled()) {
Expand All @@ -121,11 +185,12 @@ public void onMessage(ByteBuffer message) {
authorize();
session_client.getBasicRemote().sendBinary(message);
} catch (Throwable e) {
final CloseReason reason=closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"upstream message error: "+e.toString());
try {
session_client.close(new CloseReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"upstream message error: "+e.toString()));
session_client.close(reason);
} catch (Throwable e1) {}
try {
session_upstream.close(new CloseReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"upstream message error: "+e.toString()));
session_upstream.close(reason);
} catch (Throwable e1) {}
}
}
Expand All @@ -146,7 +211,7 @@ public void onClose(Session session_upstream, CloseReason reason) {
} catch (Exception e) {
logger.error("{}: {}",principal.request.getUri().asURI(),e.toString());
try {
session_client.close(new CloseReason(CloseReason.CloseCodes.TRY_AGAIN_LATER,"upstream down: "+e.toString()));
session_client.close(closeReason(CloseReason.CloseCodes.TRY_AGAIN_LATER,"upstream down: "+e.toString()));
} catch (IOException e1) {}
}
}
Expand All @@ -155,25 +220,16 @@ public void onClose(Session session_upstream, CloseReason reason) {
public void end(Session session_client,CloseReason reason) throws IOException {
logger.debug("client close {} {}: {}",session_client.getRequestURI(),session_client.getId(),reason);

try {
session_upstream.close(reason);
} catch (Throwable ioe) {}
finally {
session_upstream=null;
}
closeUpstream(reason);
}

@OnError
public void onError(Session session_client,Throwable t) throws Throwable {
// upstream first: Tomcat runs @OnClose inside close(), and end() would close the upstream normally
closeUpstream(closeReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"client error: "+t.toString()));
try {
session_client.close();
} catch (Throwable e1) {}
try {
session_upstream.close(new CloseReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"client error: "+t.toString()));
} catch (Throwable e1) {}
finally {
session_upstream=null;
}
}

@OnMessage
Expand All @@ -182,22 +238,11 @@ public void echoTextMessage(Session session_client,String msg, boolean last) {
logger.trace("->{}: {}",session_client.getRequestURI(),msg);
}
try {
int ct=0;
while ((ct++<5000)&&(session_upstream==null || !session_upstream.isOpen())) {
Thread.sleep(1);
}
authorize();
session_upstream.getBasicRemote().sendText(msg, last);
final Session upstream=upstream();
upstream.getBasicRemote().sendText(msg, last);
} catch (Throwable e) {
try {
session_client.close(new CloseReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"client message error: "+e.toString()));
} catch (Throwable e1) {}
try {
session_upstream.close(new CloseReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"client message error: "+e.toString()));
} catch (Throwable e1) {}
finally {
session_upstream=null;
}
closeAbnormally(session_client, "client message error: "+e.toString());
}
}

Expand All @@ -207,22 +252,11 @@ public void echoBinaryMessage(Session session_client,ByteBuffer bb,boolean last)
logger.trace("->{}: {}",session_client.getRequestURI(),bb.capacity());
}
try {
int ct=0;
while ((ct++<5000)&&(session_upstream==null || !session_upstream.isOpen())) {
Thread.sleep(1);
}
authorize();
session_upstream.getBasicRemote().sendBinary(bb, last);
final Session upstream=upstream();
upstream.getBasicRemote().sendBinary(bb, last);
} catch (Throwable e) {
try {
session_client.close(new CloseReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"client message error: "+e.toString()));
} catch (Throwable e1) {}
try {
session_upstream.close(new CloseReason(CloseReason.CloseCodes.CLOSED_ABNORMALLY,"client message error: "+e.toString()));
} catch (Throwable e1) {}
finally {
session_upstream=null;
}
closeAbnormally(session_client, "client message error: "+e.toString());
}
}
}
47 changes: 47 additions & 0 deletions openig-core/src/test/java/org/forgerock/openig/text/SeparatedValuesFileTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,47 @@
/*
* The contents of this file are subject to the terms of the Common Development and
* Distribution License (the License). You may not use this file except in compliance with the
* License.
*
* You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
* specific language governing permission and limitations under the License.
*
* When distributing Covered Software, include this CDDL Header Notice in each file and include
* the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
* Header, with the fields enclosed by brackets [] replaced by your own identifying
* information: "Portions copyright [year] [name of copyright owner]".
*
* Copyright 2026 3A Systems, LLC.
*/

package org.forgerock.openig.text;

import static java.nio.charset.StandardCharsets.UTF_8;
import static org.assertj.core.api.Assertions.assertThat;

import java.nio.file.Files;
import java.nio.file.Path;

import org.testng.annotations.Test;

@SuppressWarnings("javadoc")
public class SeparatedValuesFileTest {

@Test
public void shouldReadRecordFromFile() throws Exception {
Path csv = Files.createTempFile("users", ".csv");
try {
Files.write(csv, "username,password\njoe,secret\nbob,hunter2\n".getBytes(UTF_8));

SeparatedValuesFile file = new SeparatedValuesFile(csv.toFile());

assertThat(file.getRecord("username", "bob"))
.containsEntry("username", "bob")
.containsEntry("password", "hunter2");
assertThat(file.getRecord("username", "nobody")).isNull();
assertThat(file.getRecord("missing", "joe")).isNull();
} finally {
Files.delete(csv);
}
}
}
Loading
Loading