fix: fix potential deadlock in AtmospherePushConnection disconnect (#24133) (CP: 25.0) (#24158)

This PR cherry-picks changes from the original PR #24133 to branch 25.0.
---
#### Original PR description
> Fixes #24129

Co-authored-by: Marco Collovati <marco@vaadin.com>

Author: vaadin-bot

flow-server/src/main/java/com/vaadin/flow/server/communication/AtmospherePushConnection.java

@@ -23,6 +23,7 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.atmosphere.cpr.AtmosphereResource;
 import org.atmosphere.cpr.AtmosphereResource.TRANSPORT;
@@ -54,7 +55,7 @@ public class AtmospherePushConnection
     private transient FragmentedMessage incomingMessage;
     private transient Future<Object> outgoingMessage;
     private transient Object lock = new Object();
-    private volatile boolean disconnecting;
+    private AtomicBoolean disconnecting = new AtomicBoolean(false);
 
     /**
      * Represents a message that can arrive as multiple fragments.
@@ -189,8 +190,9 @@ public void push() {
      *            false if it is a response to a client request.
      */
     public void push(boolean async) {
-        if (disconnecting || !isConnected()) {
-            if (disconnecting) {
+        boolean isDisconnecting = disconnecting.get();
+        if (isDisconnecting || !isConnected()) {
+            if (isDisconnecting) {
                 getLogger().debug(
                         "Disconnection in progress, ignoring push request");
             }
@@ -325,7 +327,11 @@ public void disconnect() {
         // to skip the operation. This also prevents potential deadlocks if the
         // container acquires locks during operations on HTTP session, as
         // closing the AtmosphereResource may cause HTTP session access
-        if (disconnecting) {
+        // Atomically claim the right to disconnect. Only one thread can
+        // pass this gate - eliminates the TOCTOU (time-of-check-time-of-use)
+        // race that existed when the volatile boolean was checked outside
+        // synchronized(lock) but set inside it.
+        if (!disconnecting.compareAndSet(false, true)) {
             getLogger().debug(
                     "Disconnection already in progress, ignoring request");
             return;
@@ -340,7 +346,6 @@ public void disconnect() {
                 return;
             }
             try {
-                disconnecting = true;
                 if (resource.isResumed()) {
                     // This can happen for long polling because of
                     // http://dev.vaadin.com/ticket/16919
@@ -376,7 +381,7 @@ public void disconnect() {
                 }
                 connectionLost();
             } finally {
-                disconnecting = false;
+                disconnecting.set(false);
             }
         }
     }
@@ -435,7 +440,7 @@ private void readObject(ObjectInputStream stream)
             throws IOException, ClassNotFoundException {
         stream.defaultReadObject();
         state = State.DISCONNECTED;
-        disconnecting = false;
+        disconnecting = new AtomicBoolean(false);
         lock = new Object();
     }