Active Object Pattern in Concurrent Java 1.5+

Question

I am trying to develop active object pattern in concurrent Java using java.util.concurrent classes.

I describe it using a Client and a Server. A sample Server is as:

class Server implements Runnable {
  public final LinkedBlockingQueue que = new LinkedBlockingQueue();
  private final ExecutorService es = Executors.newCachedThreadPool();
  private Message currentMessage;

  private boolean state = false;

  public init() {
    es.submit(this);
  }

  public void requestForServer() {
    if (state) {
      this.currentMessage.await();
    }
    state = true;
  }

  public void run() {
     for(;;) {
       Message m = que.take();
       this.currentMessage = m;
       this.es.submit(m);           
     }
  }

}

And a sample Client:

class Client {

  private Server server;

  public Client(Server s) {
    this.server = s;
  }

  public void doSomething() {
    Message m = new Message(new Callable() {
      public Object call() {
        server.requestForServer();
      }
    });
    this.server.que.add(m);
  }

}

And a sample Message encapsulation is:

class Message<V> extends FutureTask<V> {
  private Lock lock = new ReentrantLock();
  private Condition condition = new Condition();

  public Message(Callable<V> callable) {
    super(callable);
  }

  public void run() {
    try {
      lock.lock();
      super.run();
      lock.unlock();
    } catch(Exception e) {}
  }

  public void await() {
    try {
      condition.await();
    } catch(Exception e) {}
  }

  public void signal() {
    try {
      condition.signalAll();
    } catch(Exception e) {}
  }

}

And a sample running code:

Server s = new Server();
Client c = new Client (s);
s.init();
c.doSomething();

I dropped some implementation details to get my message across.

Now, the problem is when in Server the state is true so the incoming message should wait and the await is called on the current message. However, I get IllegalMonitorStateException which means that the current message does not own the current thread to await on it. But, I believe this is strange since the current message gets called in the Server and its thread pool so the current message has also an access to the current thread of execution.

I'd be most thankful for any ideas or suggestions, or with a known working implementation of this pattern using java.util.concurrent. Thanks in advance.

UPDATE:
I discussed the solution I could deploy in this blog post. I hope it could help.

Answer 1

You have to actually acquire the lock when you await on its corresponding condition. Without that lock you cannot associate yourself to the condition directly. To demonstrate this:

  public void await() {
    lock.lock();
    try {
      condition.await();
    } catch(Exception e) {}
    finally{
      lock.unlock();
    }
  }

That should resolve your IllegalMonitorStateException

On a side note of correctness you should always release a lock in a try{ } finally{ } manner, you can observe what I wrote as an example. The reason for this is if an exception occurs between lock().lock(); and super.run(); lock.unlock() will never be called.