MINA: Performing synchronous write requests / read responses

Question

I'm attempting to perform a synchronous write/read in a demux-based client application with MINA 2.0 RC1, but it seems to get stuck. Here is my code:

public boolean login(final String username, final String password) {
    // block inbound messages
    session.getConfig().setUseReadOperation(true);

    // send the login request
    final LoginRequest loginRequest = new LoginRequest(username, password);
    final WriteFuture writeFuture = session.write(loginRequest);
    writeFuture.awaitUninterruptibly();

    if (writeFuture.getException() != null) {
        session.getConfig().setUseReadOperation(false);
        return false;
    }

    // retrieve the login response
    final ReadFuture readFuture = session.read();
    readFuture.awaitUninterruptibly();

    if (readFuture.getException() != null) {
        session.getConfig().setUseReadOperation(false);
        return false;
    }

    // stop blocking inbound messages
    session.getConfig().setUseReadOperation(false);

    // determine if the login info provided was valid
    final LoginResponse loginResponse = (LoginResponse)readFuture.getMessage();
    return loginResponse.getSuccess();
}

I can see on the server side that the LoginRequest object is retrieved, and a LoginResponse message is sent. On the client side, the DemuxingProtocolCodecFactory receives the response, but after throwing in some logging, I can see that the client gets stuck on the call to readFuture.awaitUninterruptibly().

I can't for the life of me figure out why it is stuck here based upon my own code. I properly set the read operation to true on the session config, meaning that messages should be blocked. However, it seems as if the message no longer exists by time I try to read response messages synchronously.

Any clues as to why this won't work for me?

Answer 1

The reason this wasn't working for me was because of an issue elsewhere in my code where I stupidly neglected to implement the message response encoder/decoder. Ugh. Anyway, the code in my question worked as soon as I fixed that.

Answer 2

I had this exact problem. It turns out that it's because I was doing reads/writes in my IoHandler.sessionCreated() implementation. I moved the processing onto the thread that established the connection, instead of just waiting for the close future.

Answer 3

I prefer this one (Christian Mueller : http://apache-mina.10907.n7.nabble.com/Mina-Client-which-sends-receives-messages-synchronous-td35672.html)

public class UCPClient { 

private Map<Integer, BlockingQueue<UCPMessageResponse>> concurrentMap = new ConcurrentHashMap<Integer, BlockingQueue<UCPMessageResponse>>(); 

// some other code 

public UCPMessageResponse send(UCPMessageRequest request) throws Throwable { 
    BlockingQueue<UCPMessageResponse> queue = new LinkedBlockingQueue<UCPMessageResponse>(1); 
    UCPMessageResponse res = null; 

    try { 
        if (sendSync) { 
            concurrentMap.put(Integer.valueOf(request.getTransactionReference()), queue); 
        } 

        WriteFuture writeFuture = session.write(request); 

        if (sendSync) { 
            boolean isSent = writeFuture.await(transactionTimeout, TimeUnit.MILLISECONDS); 

            if (!isSent) { 
                throw new TimeoutException("Could not sent the request in " + transactionTimeout + " milliseconds."); 

            } 

            if (writeFuture.getException() != null) { 
                throw writeFuture.getException(); 
            } 

            res = queue.poll(transactionTimeout, TimeUnit.MILLISECONDS); 

            if (res == null) { 
                throw new TimeoutException("Could not receive the response in " + transactionTimeout + " milliseconds."); 
            } 
        } 
    } finally { 
        if (sendSync) { 
            concurrentMap.remove(Integer.valueOf(request.getTransactionReference())); 
        } 
    } 

    return res; 
} 

}

and the IoHandler:

public class InnerHandler implements IoHandler { 

// some other code 

public void messageReceived(IoSession session, Object message) throws Exception { 
    if (sendSync) { 
        UCPMessageResponse res = (UCPMessageResponse) message; 
        BlockingQueue<UCPMessageResponse> queue = concurrentMap.get(res.getTransactionReference()); 
        queue.offer(res); 
    } 
} 

}

Answer 4

You must not use your login() function in IoHandler Thread :

If you call IoFuture.awaitUninterruptibly() in the override event function of IoHandler,

IoHandler don't work and get stuck.

You can call login() in other Thread and it will be work properly.