MINA: выполнение синхронных запросов на запись / ответов на чтение

Question

Я пытаюсь выполнить синхронную запись / чтение в клиентском приложении на основе demux с MINA 2.0 RC1, но, похоже, застрял. Вот мой код:

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();
}

Я вижу на стороне сервера, что объект LoginRequest получен и отправлено сообщение LoginResponse. На стороне клиента DemuxingProtocolCodecFactory получает ответ, но после добавления некоторой регистрации я вижу, что клиент застревает при вызове readFuture.awaitUninterruptibly().

Я не могу понять, почему он застрял здесь, основываясь на моем собственном коде. Я правильно установил для операции чтения значение true в конфигурации сеанса, что означает, что сообщения должны быть заблокированы. Однако создается впечатление, что сообщение больше не существует к тому времени, когда я пытаюсь прочитать ответные сообщения синхронно.

Есть какие-нибудь подсказки, почему это не сработает для меня?

Answer 1

Я предпочитаю это (Кристиан Мюллер: http://apache -mina.10907.n7.nabble.com / Mina-Client-which-send-receive-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; 
} 

}

и 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 2

У меня была именно эта проблема. Оказывается, это потому, что я делал чтение / запись в моей реализации IoHandler.sessionCreated (). Я перенес обработку на поток, который установил соединение, вместо того, чтобы просто ждать ближайшего будущего.

Answer 3

Вы не должны использовать свою функцию login() в IoHandler Тема:

Если вы вызываете IoFuture.awaitUninterruptibly() в функции переопределения события IoHandler,

IoHandler не работает и застревает.

Вы можете позвонить login() в другом потоке, и он будет работать правильно.

Answer 4

Причина, по которой это не сработало для меня, была из-за проблемы в другом месте моего кода, когда я тупо пренебрегал реализацией кодера / декодера ответа на сообщение. Тьфу. Во всяком случае, код в моем вопросе сработал, как только я это исправил.