[ISSUE #10860] Make DefaultPromise.get wait for completion

openmessaging/src/main/java/io/openmessaging/rocketmq/promise/DefaultPromise.java

@@ -58,7 +58,7 @@ public boolean isDone() {
 
     @Override
     public V get() {
-        return result;
+        return get(0);
     }
 
     @Override
@@ -69,10 +69,14 @@ public V get(final long timeout) {
             }
 
             if (timeout <= 0) {
-                try {
-                    lock.wait();
-                } catch (Exception e) {
-                    cancel(e);
+                while (isDoing()) {
+                    try {
+                        lock.wait();
+                    } catch (InterruptedException e) {
+                        cancel(e);
+                        Thread.currentThread().interrupt();
+                        break;
+                    }
                 }
                 return getValueOrThrowable();
             } else {
@@ -177,7 +181,6 @@ private V getValueOrThrowable() {
             Throwable e = exception.getCause() != null ? exception.getCause() : exception;
             throw new OMSRuntimeException("-1", e);
         }
-        notifyListeners();
         return result;
     }
 
@@ -222,4 +225,3 @@ private boolean cancel(Exception e) {
         return true;
     }
 }
-

openmessaging/src/test/java/io/openmessaging/rocketmq/promise/DefaultPromiseTest.java

@@ -20,6 +20,11 @@
 import io.openmessaging.FutureListener;
 import io.openmessaging.Promise;
 import io.openmessaging.exception.OMSRuntimeException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -52,6 +57,52 @@ public void testGet() throws Exception {
         assertThat(promise.get()).isEqualTo("Done");
     }
 
+    @Test
+    public void testGetWaitsForCompletion() throws Exception {
+        CountDownLatch getStarted = new CountDownLatch(1);
+        FutureTask<String> getTask = new FutureTask<>(() -> {
+            getStarted.countDown();
+            return promise.get();
+        });
+        Thread getThread = new Thread(getTask, "DefaultPromiseTestGetThread");
+        getThread.setDaemon(true);
+        getThread.start();
+
+        assertThat(getStarted.await(5, TimeUnit.SECONDS)).isTrue();
+        try {
+            getTask.get(200, TimeUnit.MILLISECONDS);
+            failBecauseExceptionWasNotThrown(TimeoutException.class);
+        } catch (TimeoutException expected) {
+            assertThat(expected).isNotNull();
+        }
+
+        promise.set("Done");
+        assertThat(getTask.get(5, TimeUnit.SECONDS)).isEqualTo("Done");
+    }
+
+    @Test
+    public void testGetPropagatesFailure() {
+        IllegalStateException failure = new IllegalStateException("Test failure");
+        promise.setFailure(failure);
+
+        try {
+            promise.get();
+            failBecauseExceptionWasNotThrown(OMSRuntimeException.class);
+        } catch (OMSRuntimeException e) {
+            assertThat(e.getCause()).isEqualTo(failure);
+        }
+    }
+
+    @Test
+    public void testGetDoesNotNotifyListenerAgain() {
+        AtomicInteger notificationCount = new AtomicInteger();
+        promise.addListener(future -> notificationCount.incrementAndGet());
+
+        promise.set("Done");
+        assertThat(promise.get()).isEqualTo("Done");
+        assertThat(notificationCount).hasValue(1);
+    }
+
     @Test
     public void testGet_WithTimeout() throws Exception {
         try {
@@ -117,4 +168,4 @@ public void getThrowable() throws Exception {
         assertThat(promise.getThrowable()).isEqualTo(exception);
     }
 
-}
+}