From 4a66e2d1ab32795c8f17776e488bce19ae43ff33 Mon Sep 17 00:00:00 2001
From: Brennan <brecon@microsoft.com>
Date: Mon, 8 Jun 2020 18:25:49 -0700
Subject: [PATCH] [Java] Observe error to avoid crash (#22016)

---
 .../signalr/OkHttpWebSocketWrapper.java       | 21 +++++++++++++++++--
 1 file changed, 19 insertions(+), 2 deletions(-)

diff --git a/src/SignalR/clients/java/signalr/src/main/java/com/microsoft/signalr/OkHttpWebSocketWrapper.java b/src/SignalR/clients/java/signalr/src/main/java/com/microsoft/signalr/OkHttpWebSocketWrapper.java
index 7dfa1706ad..b76f02fe26 100644
--- a/src/SignalR/clients/java/signalr/src/main/java/com/microsoft/signalr/OkHttpWebSocketWrapper.java
+++ b/src/SignalR/clients/java/signalr/src/main/java/com/microsoft/signalr/OkHttpWebSocketWrapper.java
@@ -4,6 +4,7 @@
 package com.microsoft.signalr;
 
 import java.util.Map;
+import java.util.concurrent.locks.ReentrantLock;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -26,6 +27,7 @@ class OkHttpWebSocketWrapper extends WebSocketWrapper {
     private WebSocketOnClosedCallback onClose;
     private CompletableSubject startSubject = CompletableSubject.create();
     private CompletableSubject closeSubject = CompletableSubject.create();
+    private final ReentrantLock closeLock = new ReentrantLock();
 
     private final Logger logger = LoggerFactory.getLogger(OkHttpWebSocketWrapper.class);
 
@@ -87,14 +89,29 @@ class OkHttpWebSocketWrapper extends WebSocketWrapper {
         @Override
         public void onClosing(WebSocket webSocket, int code, String reason) {
             onClose.invoke(code, reason);
-            closeSubject.onComplete();
+            try {
+                closeLock.lock();
+                closeSubject.onComplete();
+            }
+            finally {
+                closeLock.unlock();
+            }
             checkStartFailure();
         }
 
         @Override
         public void onFailure(WebSocket webSocket, Throwable t, Response response) {
             logger.error("WebSocket closed from an error: {}.", t.getMessage());
-            closeSubject.onError(new RuntimeException(t));
+
+            try {
+                closeLock.lock();
+                if (!closeSubject.hasComplete()) {
+                    closeSubject.onError(new RuntimeException(t));
+                }
+            }
+            finally {
+                closeLock.unlock();
+            }
             onClose.invoke(null, t.getMessage());
             checkStartFailure();
         }