=tck Improve Promise to be race-free and avoiding queue if not needed

Author: ktoso

tck/src/main/java/org/reactivestreams/tck/TestEnvironment.java

@@ -25,6 +25,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static org.testng.Assert.assertTrue;
 import static org.testng.Assert.fail;
@@ -192,7 +193,7 @@ public void flop(Throwable thr, String msg) {
       asyncErrors.add(thr);
     }
   }
-  
+
   /**
    * To flop means to "fail asynchronously", either by onErroring or by failing some TCK check triggered asynchronously.
    * This method does *NOT* fail the test - it's up to inspections of the error to fail the test if required.
@@ -886,34 +887,39 @@ public Promise(TestEnvironment env) {
     }
 
     private ArrayBlockingQueue<T> abq = new ArrayBlockingQueue<T>(1);
-    private volatile T _value = null;
+    private AtomicReference<T> _value = new AtomicReference<T>();
 
     public T value() {
-      if (isCompleted()) {
-        return _value;
+      final T value = _value.get();
+      if (value != null) {
+        return value;
       } else {
         env.flop("Cannot access promise value before completion");
         return null;
       }
     }
 
     public boolean isCompleted() {
-      return _value != null;
+      return _value.get() != null;
     }
 
     /**
      * Allows using expectCompletion to await for completion of the value and complete it _then_
      */
     public void complete(T value) {
-      abq.add(value);
+      if (_value.compareAndSet(null, value)) {
+        // we add the value to the queue such to wake up any expectCompletion which was triggered before complete() was called
+        abq.add(value);
+      }
     }
 
     /**
-     * Completes the promise right away, it is not possible to expectCompletion on a Promise completed this way
+     * Same as complete.
+     *
+     * Keeping this method for binary compatibility.
      */
     public void completeImmediatly(T value) {
-      complete(value); // complete!
-      _value = value;  // immediatly!
+      complete(value);
     }
 
     public void expectCompletion(long timeoutMillis, String errorMsg) throws InterruptedException {
@@ -922,8 +928,6 @@ public void expectCompletion(long timeoutMillis, String errorMsg) throws Interru
 
         if (val == null) {
           env.flop(String.format("%s within %d ms", errorMsg, timeoutMillis));
-        } else {
-          _value = val;
         }
       }
     }