[cxf] branch master updated: Fixing SseEventSource isOpen() semantics (addressing JAX-RS TCK failures) (#796)

[reta]

This is an automated email from the ASF dual-hosted git repository.

reta pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/cxf.git


The following commit(s) were added to refs/heads/master by this push:
     new 9b92cf1  Fixing SseEventSource isOpen() semantics (addressing JAX-RS 
TCK failures) (#796)
9b92cf1 is described below

commit 9b92cf15a13c2cc90f50d8fb9be76fabe533b94a
Author: Andriy Redko <drr...@gmail.com>
AuthorDate: Fri May 14 21:37:51 2021 -0400

    Fixing SseEventSource isOpen() semantics (addressing JAX-RS TCK failures) 
(#796)
---
 .../cxf/jaxrs/sse/client/SseEventSourceImpl.java   | 38 ++++++++++++++++++----
 1 file changed, 31 insertions(+), 7 deletions(-)

diff --git 
a/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/client/SseEventSourceImpl.java
 
b/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/client/SseEventSourceImpl.java
index 4e8ede2..c823e3c 100644
--- 
a/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/client/SseEventSourceImpl.java
+++ 
b/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/client/SseEventSourceImpl.java
@@ -59,6 +59,7 @@ public class SseEventSourceImpl implements SseEventSource {
     private volatile InboundSseEventProcessor processor; 
     private volatile TimeUnit unit;
     private volatile long delay;
+    private volatile boolean open;
 
     private class InboundSseEventListenerDelegate implements 
InboundSseEventListener {
         private String lastEventId;
@@ -163,7 +164,7 @@ public class SseEventSourceImpl implements SseEventSource {
 
     @Override
     public void open() {
-        if (!state.compareAndSet(SseSourceState.CLOSED, 
SseSourceState.CONNECTING)) {
+        if (!tryChangeState(SseSourceState.CLOSED, SseSourceState.CONNECTING)) 
{
             throw new IllegalStateException("The SseEventSource is already in 
" + state.get() + " state");
         }
 
@@ -199,7 +200,7 @@ public class SseEventSourceImpl implements SseEventSource {
             final int status = response.getStatus();
             if (status == 204) {
                 LOG.fine("SSE endpoint " + target.getUri() + " returns no 
data, disconnecting");
-                state.set(SseSourceState.CLOSED);
+                changeState(SseSourceState.CLOSED);
                 response.close();
                 return;
             }
@@ -229,7 +230,7 @@ public class SseEventSourceImpl implements SseEventSource {
             processor.run(response);
             LOG.fine("SSE event processor has been started ...");
             
-            if (!state.compareAndSet(SseSourceState.CONNECTING, 
SseSourceState.OPEN)) {
+            if (!tryChangeState(SseSourceState.CONNECTING, 
SseSourceState.OPEN)) {
                 throw new IllegalStateException("The SseEventSource is already 
in " + state.get() + " state");
             }
             
@@ -252,7 +253,7 @@ public class SseEventSourceImpl implements SseEventSource {
 
     @Override
     public boolean isOpen() {
-        return state.get() == SseSourceState.OPEN;
+        return open;
     }
 
     @Override
@@ -261,9 +262,9 @@ public class SseEventSourceImpl implements SseEventSource {
             return true;
         }
         
-        if (state.compareAndSet(SseSourceState.CONNECTING, 
SseSourceState.CLOSED)) {
+        if (tryChangeState(SseSourceState.CONNECTING, SseSourceState.CLOSED)) {
             LOG.fine("The SseEventSource was not connected, closing anyway");
-        } else if (!state.compareAndSet(SseSourceState.OPEN, 
SseSourceState.CLOSED)) {
+        } else if (!tryChangeState(SseSourceState.OPEN, 
SseSourceState.CLOSED)) {
             throw new IllegalStateException("The SseEventSource is not opened, 
but in " + state.get() + " state");
         }
         
@@ -299,7 +300,7 @@ public class SseEventSourceImpl implements SseEventSource {
         // If the connection was still on connecting state, just try to 
reconnect
         if (state.get() != SseSourceState.CONNECTING) { 
             LOG.fine("The SseEventSource is still opened, moving it to 
connecting state");
-            if (!state.compareAndSet(SseSourceState.OPEN, 
SseSourceState.CONNECTING)) {
+            if (!tryChangeState(SseSourceState.OPEN, 
SseSourceState.CONNECTING)) {
                 throw new IllegalStateException("The SseEventSource is not 
opened, but in " + state.get()
                     + " state, unable to reconnect");
             }
@@ -316,4 +317,27 @@ public class SseEventSourceImpl implements SseEventSource {
         LOG.fine("The reconnection attempt to " + target.getUri() + " is 
scheduled in "
             + tunit.toMillis(tdelay) + "ms");
     }
+    
+    private void changeState(SseSourceState updated) {
+        state.set(updated);
+        onStateChanged(updated);
+    }
+    
+    private boolean tryChangeState(SseSourceState expected, SseSourceState 
updated) {
+        final boolean result = state.compareAndSet(expected, updated);
+        
+        if (result) {
+            onStateChanged(updated);
+        }
+        
+        return result;
+    }
+
+    private void onStateChanged(SseSourceState updated) {
+        if (state.get() == SseSourceState.OPEN) {
+            open = true;
+        } else if (state.get() == SseSourceState.CLOSED) {
+            open = false;
+        }
+    }
 }