2
\$\begingroup\$

I have a system that can put and get messages from a messaging middleware (JMS), which is asynchronous. I need to provide an API that allows synchronous call, as in, when the API/method is called, it returns only when a response message is received or until a specified time out. In short, the calling thread needs to wait until woken up when the response message is received or until the time out.

CompletableFuture from Java 8 does the job, but this library needs to work in Java 7 and I couldn't find any equivalent in Java 7. Guava has ListenableFuture, but it doesn't fit my requirement as I need the thread that called to be held while waiting for the response.

Here is a class that can hold the calling thread and return when a response is delivered by another thread. I am looking to avoid any code smells or concurrency issues. Any additional tests and constructive comments are welcome!

import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Object to hold the calling thread while we wait for a response message from client
 *
 */
public class SyncResult<T> {

    private T result;

    private final ReentrantLock messageLock = new ReentrantLock();
    private final Condition messageCondition = messageLock.newCondition();
    // In the very remote off chance that result is already available and 
    // wait is not required
    private final AtomicBoolean messageReceived = new AtomicBoolean(false);

    /**
     * Sets the given result and notifies the calling thread to return the result
     */
    public void setResult(T result) {
        messageLock.lock();
        this.result = result;
        messageReceived.set(true);
        messageCondition.signal();
        messageLock.unlock();
    }

    /**
     * Return the result or time out and get an exception
     */ 
    public T getResult(long timeOut, TimeUnit timeUnit) throws InterruptedException,
        TimeoutException {
        boolean receivedSignal = false;
        messageLock.lock();
        if(!messageReceived.get()) {
            receivedSignal = messageCondition.await(timeOut, timeUnit);
        } 
        messageLock.unlock();
        // If messageReceived is true, then we return null
        if(!receivedSignal && !messageReceived.get()) {
            throw new TimeoutException();
        }
        return result;
    }
}

Unit tests:

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.Assert;
import org.testng.annotations.Test;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class SyncResultTests {

    private static final Logger LOGGER = LoggerFactory.getLogger(SyncResultTests.class);

    @Test
    public void testTimelyDelivery() throws InterruptedException, TimeoutException {
        final SyncResult<String> syncResult = new SyncResult<>();
        final String myResult = "Done";

        Runnable messageWorker = new Runnable() {
            @Override
            public void run() {
                syncResult.setResult(myResult);
            }
        };

        ExecutorService executors = Executors.newSingleThreadExecutor();
        executors.submit(messageWorker);

        String result = syncResult.getResult(5, TimeUnit.SECONDS);
        Assert.assertEquals(myResult, result);
    }

    @Test
    public void testEarlyDelivery() throws InterruptedException, TimeoutException {
        final SyncResult<String> syncResult = new SyncResult<>();
        final String myResult = "Done";

        Runnable messageWorker = new Runnable() {
            @Override
            public void run() {
                syncResult.setResult(myResult);
            }
        };

        ExecutorService executors = Executors.newSingleThreadExecutor();
        executors.submit(messageWorker);
        executors.shutdown();
        Assert.assertTrue(executors.awaitTermination(5, TimeUnit.SECONDS), "Message result setter thread not terminated");

        long startTime = System.currentTimeMillis();
        String result = syncResult.getResult(5, TimeUnit.SECONDS);
        long elapsedTime = System.currentTimeMillis() - startTime;
        Assert.assertEquals(myResult, result);
        Assert.assertTrue(elapsedTime < 2000L);

    }

    @Test
    public void testLateDelivery() throws InterruptedException, TimeoutException {
        final SyncResult<String> syncResult = new SyncResult<>();
        final String myResult = "Done";

        Runnable messageWorker = new Runnable() {
            @Override
            public void run() {
                syncResult.setResult(myResult);
            }
        };

        try {
            syncResult.getResult(2, TimeUnit.SECONDS);
            Assert.fail("Result should not be returned");
        } catch (TimeoutException e) {
            // Do nothing
        }

        // We should still be able to deliver a result
        ExecutorService executors = Executors.newSingleThreadExecutor();
        executors.submit(messageWorker);
    }

    @Test(expectedExceptions = TimeoutException.class)
    public void testNoDelivery() throws InterruptedException, TimeoutException {
        final SyncResult<String> syncResult = new SyncResult<>();
        syncResult.getResult(2, TimeUnit.SECONDS);
        Assert.fail("Result should not be returned");
    }
}
\$\endgroup\$
2
  • \$\begingroup\$ What is the reason for testing result == null? \$\endgroup\$
    – vnp
    Commented Dec 24, 2016 at 23:53
  • \$\begingroup\$ Good point. My first draft did not have the messageReceived field, so I relied on result == null. I just updated it to use a better condition, the boolean value from Condition.await(). Thanks! \$\endgroup\$
    – Sundeep
    Commented Dec 25, 2016 at 0:04

1 Answer 1

2
\$\begingroup\$

here are my suggestions and the full code afterward:

  1. Always unlock in a finally clause. This is specially true in the getResult method, if the await call is interrupted, the lock will never be released.

  2. Use signalAll instead of signal, if several threads are waiting for the result, with signal only one will stop waiting.

  3. When declaring a variable, use the most generic type. In your code, you should declare messageLock as a Lock instead of a ReentrantLock.

  4. the messageReceived variable is always used in a lock/unlock block and then could be a simple boolean instead of an AtomicBoolean

Below is the code with these suggestions :

import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Object to hold the calling thread while we wait for a response message from client
 *
 */
public class SyncResult<T> {

    private T result;

    private final Lock messageLock = new ReentrantLock();
    private final Condition messageCondition = messageLock.newCondition();
    // In the very remote off chance that result is already available and 
    // wait is not required
    private boolean messageReceived = false;

    /**
     * Sets the given result and notifies the calling thread to return the result
     */
    public void setResult(T result) {
        messageLock.lock();
        try {
            this.result = result;
            messageReceived = true;
            messageCondition.signalAll();
        } finally {
            messageLock.unlock();
        }
    }

    /**
     * Return the result or time out and get an exception
     */ 
    public T getResult(long timeOut, TimeUnit timeUnit) throws InterruptedException,
        TimeoutException {
        boolean receivedSignal = false;
        messageLock.lock();
        try {
            if(!messageReceived) {
                receivedSignal = messageCondition.await(timeOut, timeUnit);
            } 
            // If messageReceived is true, then we return null
            if(!receivedSignal && !messageReceived) {
                throw new TimeoutException();
            }
            return result;
        }
        finally {
            messageLock.unlock();
        }
    }
}
\$\endgroup\$
3
  • \$\begingroup\$ #2 is a great point I missed as I was intending for only one thread to call this object, but it can easily be called by multiple threads. Excellent catch on #4 as well, as the variables in the lock block will be synchronized among threads. However, the messageReceived field cannot be final in the code above. \$\endgroup\$
    – Sundeep
    Commented Dec 27, 2016 at 15:51
  • \$\begingroup\$ #4 yep you are right, I modified the answer accordingly. \$\endgroup\$ Commented Dec 27, 2016 at 15:56
  • \$\begingroup\$ Welcome to Code Review, this is a well formed answer! \$\endgroup\$
    – Malachi
    Commented Dec 27, 2016 at 16:28

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.