@Test public void testWholeResponse() throws Exception { Request req = new Request(1, new Object(), Long.MAX_VALUE); Response res = new Response(req, new Message(1, ResponseType.WHOLE), ResponseType.WHOLE); WriteRequest rwr = new DefaultWriteRequest(req); nextFilter.filterWrite(session, new DefaultWriteRequest(req .getMessage())); nextFilterControl.setMatcher(matcher); nextFilter.messageSent(session, rwr); nextFilter.messageReceived(session, res); nextFilterControl.replay(); filter.filterWrite(nextFilter, session, rwr); filter.messageSent(nextFilter, session, matcher.getLastWriteRequest()); filter.messageReceived(nextFilter, session, res.getMessage()); filter.messageReceived(nextFilter, session, res.getMessage()); nextFilterControl.verify(); Assert.assertEquals(res, req.awaitResponse()); assertNoSuchElementException(req); }
@Override public void messageReceived(NextFilter nextFilter, IoSession session, Object message) throws Exception { ResponseInspector responseInspector = (ResponseInspector) session .getAttribute(RESPONSE_INSPECTOR); Object requestId = responseInspector.getRequestId(message); if (requestId == null) { nextFilter.messageReceived(session, message); return; } ResponseType type = responseInspector.getResponseType(message); if (type == null) { nextFilter.exceptionCaught(session, new IllegalStateException( responseInspector.getClass().getName() + "#getResponseType() may not return null.")); } Map<Object, Request> requestStore = getRequestStore(session); Request request; switch (type) { case WHOLE: case PARTIAL_LAST: synchronized (requestStore) { request = requestStore.remove(requestId); } break; case PARTIAL: synchronized (requestStore) { request = requestStore.get(requestId); } break; default: throw new InternalError(); } if (request == null) { if (LOGGER.isWarnEnabled()) { LOGGER.warn("Unknown request ID '" + requestId + "' for the response message. Timed out already?: " + message); } } else { if (type != ResponseType.PARTIAL) { ScheduledFuture<?> scheduledFuture = request.getTimeoutFuture(); if (scheduledFuture != null) { scheduledFuture.cancel(false); Set<Request> unrespondedRequests = getUnrespondedRequestStore(session); synchronized (unrespondedRequests) { unrespondedRequests.remove(request); } } } Response response = new Response(request, message, type); request.signal(response); nextFilter.messageReceived(session, response); } }
RequestResponseFilter extends WriteRequestFilter { @Override public void messageReceived(NextFilter nextFilter, IoSession session, Object message) throws Exception { ResponseInspector responseInspector = (ResponseInspector) session .getAttribute(RESPONSE_INSPECTOR); Object requestId = responseInspector.getRequestId(message); if (requestId == null) { nextFilter.messageReceived(session, message); return; } ResponseType type = responseInspector.getResponseType(message); if (type == null) { nextFilter.exceptionCaught(session, new IllegalStateException( responseInspector.getClass().getName() + "#getResponseType() may not return null.")); } Map<Object, Request> requestStore = getRequestStore(session); Request request; switch (type) { case WHOLE: case PARTIAL_LAST: synchronized (requestStore) { request = requestStore.remove(requestId); } break; case PARTIAL: synchronized (requestStore) { request = requestStore.get(requestId); } break; default: throw new InternalError(); } if (request == null) { if (LOGGER.isWarnEnabled()) { LOGGER.warn("Unknown request ID '" + requestId + "' for the response message. Timed out already?: " + message); } } else { if (type != ResponseType.PARTIAL) { ScheduledFuture<?> scheduledFuture = request.getTimeoutFuture(); if (scheduledFuture != null) { scheduledFuture.cancel(false); Set<Request> unrespondedRequests = getUnrespondedRequestStore(session); synchronized (unrespondedRequests) { unrespondedRequests.remove(request); } } } Response response = new Response(request, message, type); request.signal(response); nextFilter.messageReceived(session, response); } } }
RequestResponseFilter extends WriteRequestFilter { @Override public void messageReceived(NextFilter nextFilter, IoSession session, Object message) throws Exception { ResponseInspector responseInspector = (ResponseInspector) session .getAttribute(RESPONSE_INSPECTOR); Object requestId = responseInspector.getRequestId(message); if (requestId == null) { nextFilter.messageReceived(session, message); return; } ResponseType type = responseInspector.getResponseType(message); if (type == null) { nextFilter.exceptionCaught(session, new IllegalStateException( responseInspector.getClass().getName() + "#getResponseType() may not return null.")); } Map<Object, Request> requestStore = getRequestStore(session); Request request; switch (type) { case WHOLE: case PARTIAL_LAST: synchronized (requestStore) { request = requestStore.remove(requestId); } break; case PARTIAL: synchronized (requestStore) { request = requestStore.get(requestId); } break; default: throw new InternalError(); } if (request == null) { if (LOGGER.isWarnEnabled()) { LOGGER.warn("Unknown request ID '" + requestId + "' for the response message. Timed out already?: " + message); } } else { if (type != ResponseType.PARTIAL) { ScheduledFuture<?> scheduledFuture = request.getTimeoutFuture(); if (scheduledFuture != null) { scheduledFuture.cancel(false); Set<Request> unrespondedRequests = getUnrespondedRequestStore(session); synchronized (unrespondedRequests) { unrespondedRequests.remove(request); } } } Response response = new Response(request, message, type); request.signal(response); nextFilter.messageReceived(session, response); } } RequestResponseFilter(final ResponseInspector responseInspector,
            ScheduledExecutorService timeoutScheduler);  RequestResponseFilter(
            ResponseInspectorFactory responseInspectorFactory,
            ScheduledExecutorService timeoutScheduler); }
RequestResponseFilter extends WriteRequestFilter { @Override public void messageReceived(NextFilter nextFilter, IoSession session, Object message) throws Exception { ResponseInspector responseInspector = (ResponseInspector) session .getAttribute(RESPONSE_INSPECTOR); Object requestId = responseInspector.getRequestId(message); if (requestId == null) { nextFilter.messageReceived(session, message); return; } ResponseType type = responseInspector.getResponseType(message); if (type == null) { nextFilter.exceptionCaught(session, new IllegalStateException( responseInspector.getClass().getName() + "#getResponseType() may not return null.")); } Map<Object, Request> requestStore = getRequestStore(session); Request request; switch (type) { case WHOLE: case PARTIAL_LAST: synchronized (requestStore) { request = requestStore.remove(requestId); } break; case PARTIAL: synchronized (requestStore) { request = requestStore.get(requestId); } break; default: throw new InternalError(); } if (request == null) { if (LOGGER.isWarnEnabled()) { LOGGER.warn("Unknown request ID '" + requestId + "' for the response message. Timed out already?: " + message); } } else { if (type != ResponseType.PARTIAL) { ScheduledFuture<?> scheduledFuture = request.getTimeoutFuture(); if (scheduledFuture != null) { scheduledFuture.cancel(false); Set<Request> unrespondedRequests = getUnrespondedRequestStore(session); synchronized (unrespondedRequests) { unrespondedRequests.remove(request); } } } Response response = new Response(request, message, type); request.signal(response); nextFilter.messageReceived(session, response); } } RequestResponseFilter(final ResponseInspector responseInspector,
            ScheduledExecutorService timeoutScheduler);  RequestResponseFilter(
            ResponseInspectorFactory responseInspectorFactory,
            ScheduledExecutorService timeoutScheduler); @Override void onPreAdd(IoFilterChain parent, String name,
            NextFilter nextFilter); @Override void onPostRemove(IoFilterChain parent, String name,
            NextFilter nextFilter); @Override void messageReceived(NextFilter nextFilter, IoSession session,
            Object message); @Override void sessionClosed(NextFilter nextFilter, IoSession session); }
RequestResponseFilter extends WriteRequestFilter { @Override public void messageReceived(NextFilter nextFilter, IoSession session, Object message) throws Exception { ResponseInspector responseInspector = (ResponseInspector) session .getAttribute(RESPONSE_INSPECTOR); Object requestId = responseInspector.getRequestId(message); if (requestId == null) { nextFilter.messageReceived(session, message); return; } ResponseType type = responseInspector.getResponseType(message); if (type == null) { nextFilter.exceptionCaught(session, new IllegalStateException( responseInspector.getClass().getName() + "#getResponseType() may not return null.")); } Map<Object, Request> requestStore = getRequestStore(session); Request request; switch (type) { case WHOLE: case PARTIAL_LAST: synchronized (requestStore) { request = requestStore.remove(requestId); } break; case PARTIAL: synchronized (requestStore) { request = requestStore.get(requestId); } break; default: throw new InternalError(); } if (request == null) { if (LOGGER.isWarnEnabled()) { LOGGER.warn("Unknown request ID '" + requestId + "' for the response message. Timed out already?: " + message); } } else { if (type != ResponseType.PARTIAL) { ScheduledFuture<?> scheduledFuture = request.getTimeoutFuture(); if (scheduledFuture != null) { scheduledFuture.cancel(false); Set<Request> unrespondedRequests = getUnrespondedRequestStore(session); synchronized (unrespondedRequests) { unrespondedRequests.remove(request); } } } Response response = new Response(request, message, type); request.signal(response); nextFilter.messageReceived(session, response); } } RequestResponseFilter(final ResponseInspector responseInspector,
            ScheduledExecutorService timeoutScheduler);  RequestResponseFilter(
            ResponseInspectorFactory responseInspectorFactory,
            ScheduledExecutorService timeoutScheduler); @Override void onPreAdd(IoFilterChain parent, String name,
            NextFilter nextFilter); @Override void onPostRemove(IoFilterChain parent, String name,
            NextFilter nextFilter); @Override void messageReceived(NextFilter nextFilter, IoSession session,
            Object message); @Override void sessionClosed(NextFilter nextFilter, IoSession session);  }