Expect the delegate endpoint to be stored under the thread id [IMMUTA…

…NT-554] We can't use a static key for the endpoint in the config's userProperties, because that map is shared for all requests, which leads to getting the wrong endpoint under the slightest load. We can use the thread id here because onOpen will be triggered on the same thread as modifyHandshake, where we store the endpoint.
projectodd · Apr 10, 2015 · 74beb43 · 74beb43
1 parent 30c4a21
commit 74beb43
Showing 1 changed file with 14 additions and 6 deletions.
diff --git a/.../src/main/java/org/projectodd/wunderboss/web/async/websocket/DelegatingJavaxEndpoint.java b/.../src/main/java/org/projectodd/wunderboss/web/async/websocket/DelegatingJavaxEndpoint.java
@@ -23,11 +23,14 @@
 import java.io.IOException;
 
 public class DelegatingJavaxEndpoint extends Endpoint {
-    public static final String ENDPOINT_KEY = "session-endpoint";
+
+    public static String endpointKey() {
+        return "session-endpoint-" + Thread.currentThread().getId();
+    }
 
     @Override
     public void onOpen(Session session, EndpointConfig endpointConfig) {
-        Endpoint endpoint = sessionEndpoint(session);
+        Endpoint endpoint = delegate(session);
         if (endpoint != null) {
             endpoint.onOpen(session, endpointConfig);
         } else {
@@ -37,15 +40,15 @@ public void onOpen(Session session, EndpointConfig endpointConfig) {
 
     @Override
     public void onClose(Session session, CloseReason closeReason) {
-        Endpoint endpoint = sessionEndpoint(session);
+        Endpoint endpoint = delegate(session);
         if (endpoint != null) {
             endpoint.onClose(session, closeReason);
         }
     }
 
     @Override
     public void onError(Session session, Throwable err) {
-        Endpoint endpoint = sessionEndpoint(session);
+        Endpoint endpoint = delegate(session);
         if (endpoint != null) {
             endpoint.onError(session, err);
         } else {
@@ -61,10 +64,15 @@ private void close(Session session) {
         }
     }
 
-    private Endpoint sessionEndpoint(final Session session) {
-        return (Endpoint)session.getUserProperties().get(ENDPOINT_KEY);
+    private synchronized Endpoint delegate(final Session session) {
+        if (this.delegate == null) {
+            this.delegate = (Endpoint)session.getUserProperties().remove(endpointKey());
+        }
+        return this.delegate;
     }
 
+    private Endpoint delegate;
+
     private final static CloseReason POLICY_CLOSE = new CloseReason(new CloseReason.CloseCode() {
         @Override
         public int getCode() {