diff --git a/openmessaging/src/main/java/io/openmessaging/rocketmq/promise/DefaultPromise.java b/openmessaging/src/main/java/io/openmessaging/rocketmq/promise/DefaultPromise.java index 46e607a5802..6bf732ff624 100644 --- a/openmessaging/src/main/java/io/openmessaging/rocketmq/promise/DefaultPromise.java +++ b/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; } } - diff --git a/openmessaging/src/test/java/io/openmessaging/rocketmq/promise/DefaultPromiseTest.java b/openmessaging/src/test/java/io/openmessaging/rocketmq/promise/DefaultPromiseTest.java index f226edef0a2..4893011f940 100644 --- a/openmessaging/src/test/java/io/openmessaging/rocketmq/promise/DefaultPromiseTest.java +++ b/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 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); } -} \ No newline at end of file +}