org.springframework.web.context.request.async.StandardServletAsyncWebRequest#javax.servlet.AsyncEvent源码实例Demo

@Override
public void dispatch(ServletContext context, String path) {
    final HttpServletRequest httpRequest = (HttpServletRequest) servletRequest;
    httpRequest.setAttribute(ASYNC_CONTEXT_PATH, httpRequest.getContextPath());
    httpRequest.setAttribute(ASYNC_PATH_INFO, httpRequest.getPathInfo());
    httpRequest.setAttribute(ASYNC_QUERY_STRING, httpRequest.getQueryString());
    httpRequest.setAttribute(ASYNC_REQUEST_URI, httpRequest.getRequestURI());
    httpRequest.setAttribute(ASYNC_SERVLET_PATH, httpRequest.getServletPath());
    final NettyRequestDispatcher dispatcher = (NettyRequestDispatcher) context.getRequestDispatcher(path);
    ctx.executor().submit(() -> {
        try {
            dispatcher.dispatch(httpRequest, servletResponse);
            // TODO is this right?
            for (AsyncListener listener : ImmutableList.copyOf(listeners)) {
                listener.onComplete(new AsyncEvent(NettyAsyncContext.this));
            }
        } catch (ServletException | IOException e) {
            // TODO notify listeners
            e.printStackTrace();
        }
    });
}
 

@Override
public void onTimeout(AsyncEvent event) throws IOException {
    // Need to handle timeouts for THREAD_AFTER_EXIT in the listener to
    // avoid concurrency issues.
    if (endTiming == EndTiming.THREAD_AFTER_EXIT) {
        switch (asyncEnd) {
            case COMPLETE: {
                asyncContext.complete();
                break;
            }
            case DISPATCH: {
                dispatch = true;
                asyncContext.dispatch();
                break;
            }
            default:
                // NO-OP
        }
    }
    if (servletRequest.isAsyncStarted() == asyncEnd.isNone()) {
        failed.set(false);
    }
    endLatch.countDown();
}
 

public OpenEJBAsyncContext(final HttpServletRequest request, final ServletResponse response, final String contextPath) {
    if (es == null) {
        synchronized (OpenEJBAsyncContext.class) { // we don't care since impl is not really thread safe, just here for testing
            if (es == null) {
                init();
            }
        }
    }

    this.request = request;
    this.contextPath = contextPath;
    if (contextPath != null) {
        for (final AppContext app : SystemInstance.get().getComponent(ContainerSystem.class).getAppContexts()) {
            for (final WebContext web : app.getWebContexts()) {
                if (web.getContextRoot().replace("/", "").equals(contextPath.replace("/", ""))) {
                    this.context = web;
                    break;
                }
            }
        }
    }
    this.response = response;
    this.socket = Socket.class.cast(request.getAttribute("openejb_socket"));
    this.event = new AsyncEvent(this, request, response);
    INITIALIZED.add(this);
}
 

@Override
public void onTimeout(AsyncEvent event) throws IOException {
    boolean expectedAsyncStarted = true;

    TestAsyncContextImpl.track("onTimeout-");
    if (completeOnTimeout){
        event.getAsyncContext().complete();
        expectedAsyncStarted = false;
    }
    if (dispatchUrl != null) {
        event.getAsyncContext().dispatch(dispatchUrl);
        expectedAsyncStarted = false;
    }

    ServletRequest req = event.getSuppliedRequest();
    asyncStartedCorrect = (expectedAsyncStarted == req.isAsyncStarted());
}
 

@Override
public void onTimeout(AsyncEvent asyncEvent) throws IOException {
    if (isDebug) {
        logger.debug("Timeout asynchronous operation. event={}", asyncEvent);
    }

    if (asyncEvent == null) {
        if (isDebug) {
            logger.debug("Invalid event. event is null");
        }
        return;
    }

    try {
        this.asyncListenerInterceptorHelper.timeout(asyncEvent.getThrowable());
    } catch (Throwable t) {
        logger.info("Failed to async event handle. event={}", asyncEvent, t);
    }
}
 

@Test
public void startDeferredResultProcessingTimeoutAndResumeThroughCallback() throws Exception {

	final DeferredResult<Integer> deferredResult = new DeferredResult<>();
	deferredResult.onTimeout(new Runnable() {
		@Override
		public void run() {
			deferredResult.setResult(23);
		}
	});

	this.asyncManager.startDeferredResultProcessing(deferredResult);

	AsyncEvent event = null;
	this.asyncWebRequest.onTimeout(event);

	assertTrue(this.asyncManager.hasConcurrentResult());
	assertEquals(23, this.asyncManager.getConcurrentResult());
	assertEquals("/test", ((MockAsyncContext) this.servletRequest.getAsyncContext()).getDispatchedPath());
}
 

@Override
public void onComplete(AsyncEvent asyncEvent) throws IOException {
    if (isDebug) {
        logger.debug("Complete asynchronous operation. event={}", asyncEvent);
    }

    if (asyncEvent == null) {
        if (isDebug) {
            logger.debug("Invalid event. event is null");
        }
        return;
    }

    try {
        final int statusCode = getStatusCode(asyncEvent);
        this.asyncListenerInterceptorHelper.complete(asyncEvent.getThrowable(), statusCode);
    } catch (Throwable t) {
        if (isInfo) {
            logger.info("Failed to async event handle. event={}", asyncEvent, t);
        }
    }
}
 

@Test
public void startDeferredResultProcessingAfterTimeoutException() throws Exception {

	DeferredResult<Integer> deferredResult = new DeferredResult<>();
	final Exception exception = new Exception();

	DeferredResultProcessingInterceptor interceptor = new DeferredResultProcessingInterceptor() {
		@Override
		public <T> boolean handleTimeout(NativeWebRequest request, DeferredResult<T> result) throws Exception {
			throw exception;
		}
	};

	this.asyncManager.registerDeferredResultInterceptor("interceptor", interceptor);
	this.asyncManager.startDeferredResultProcessing(deferredResult);

	AsyncEvent event = null;
	this.asyncWebRequest.onTimeout(event);

	assertTrue(this.asyncManager.hasConcurrentResult());
	assertEquals(exception, this.asyncManager.getConcurrentResult());
	assertEquals("/test", ((MockAsyncContext) this.servletRequest.getAsyncContext()).getDispatchedPath());
}
 

@Override
public void onError(AsyncEvent asyncEvent) throws IOException {
    if (isDebug) {
        logger.debug("Error asynchronous operation. event={}", asyncEvent);
    }

    if (asyncEvent == null) {
        if (isInfo) {
            logger.info("Invalid event. event is null");
        }
        return;
    }

    try {
        this.asyncListenerInterceptorHelper.error(asyncEvent.getThrowable());
    } catch (Throwable t) {
        if (isInfo) {
            logger.info("Failed to async event handle. event={}", asyncEvent, t);
        }
    }
}
 

private void onAsyncComplete() {
    final boolean setupRequired = SecurityActions.currentServletRequestContext() == null;
    servletRequestContext.getCurrentServletContext().invokeRunnable(servletRequestContext.getExchange(), new Runnable() {

        @Override
        public void run() {
            //now run request listeners
            setupRequestContext(setupRequired);
            try {
                for (final BoundAsyncListener listener : asyncListeners) {
                    AsyncEvent event = new AsyncEvent(AsyncContextImpl.this, listener.servletRequest, listener.servletResponse);
                    try {
                        listener.asyncListener.onComplete(event);
                    } catch (IOException e) {
                        UndertowServletLogger.REQUEST_LOGGER.ioExceptionDispatchingAsyncEvent(e);
                    } catch (Throwable t) {
                        UndertowServletLogger.REQUEST_LOGGER.failureDispatchingAsyncEvent(t);
                    }
                }
            } finally {
                tearDownRequestContext(setupRequired);
            }
        }
    });
}
 

@Test
public void startCallableProcessingAfterException() throws Exception {

	StubCallable callable = new StubCallable();
	Exception exception = new Exception();

	CallableProcessingInterceptor interceptor = mock(CallableProcessingInterceptor.class);
	Exception e = new Exception();
	given(interceptor.handleError(this.asyncWebRequest, callable, e)).willThrow(exception);

	this.asyncManager.registerCallableInterceptor("errorInterceptor", interceptor);
	this.asyncManager.startCallableProcessing(callable);

	AsyncEvent event = new AsyncEvent(new MockAsyncContext(this.servletRequest, this.servletResponse), e);
	this.asyncWebRequest.onError(event);

	assertTrue(this.asyncManager.hasConcurrentResult());
	assertEquals(exception, this.asyncManager.getConcurrentResult());
	assertEquals("/test", ((MockAsyncContext) this.servletRequest.getAsyncContext()).getDispatchedPath());

	verify(interceptor).beforeConcurrentHandling(this.asyncWebRequest, callable);
}
 

@Override
public void onComplete(AsyncEvent event) {
  ServletResponse response = event.getSuppliedResponse();
  if (response instanceof HttpServletResponse) {
    OcHttpServletUtil.recordMessageSentEvent(handler, context, (HttpServletResponse) response);
  }
  handler.handleEnd(
      context,
      (HttpServletRequest) event.getSuppliedRequest(),
      (HttpServletResponse) event.getSuppliedResponse(),
      null);
  this.close();
}
 

private AsyncEvent customizeEvent(AsyncEvent event) {
    if (servletRequest != null && servletResponse != null) {
        return new AsyncEvent(event.getAsyncContext(), servletRequest, servletResponse,
                event.getThrowable());
    } else {
        return event;
    }
}
 

public void setStarted(Context context, ServletRequest request,
        ServletResponse response, boolean originalRequestResponse) {

    synchronized (asyncContextLock) {
        this.request.getCoyoteRequest().action(
                ActionCode.ASYNC_START, this);

        this.context = context;
        context.incrementInProgressAsyncCount();
        this.servletRequest = request;
        this.servletResponse = response;
        this.hasOriginalRequestAndResponse = originalRequestResponse;
        this.event = new AsyncEvent(this, request, response);

        List<AsyncListenerWrapper> listenersCopy = new ArrayList<>();
        listenersCopy.addAll(listeners);
        listeners.clear();
        if (log.isDebugEnabled()) {
            log.debug(sm.getString("asyncContextImpl.fireOnStartAsync"));
        }
        for (AsyncListenerWrapper listener : listenersCopy) {
            try {
                listener.fireOnStartAsync(event);
            } catch (Throwable t) {
                ExceptionUtils.handleThrowable(t);
                log.warn(sm.getString("asyncContextImpl.onStartAsyncError",
                        listener.getClass().getName()), t);
            }
        }
    }
}
 

public void onError(final Throwable throwable) {
    final AsyncEvent event = new AsyncEvent(this, request, response, throwable);
    executeOnListeners(l -> l.onError(event), null);
    if (!response.isCommitted() && HttpServletResponse.class.isInstance(response)) {
        final HttpServletResponse http = HttpServletResponse.class.cast(response);
        http.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
    }
    complete();
}
 

@Override
public void onError(AsyncEvent event) throws IOException {
    boolean expectedAsyncStarted = true;

    TestAsyncContextImpl.track("onError-");
    if (completeOnError) {
        event.getAsyncContext().complete();
        expectedAsyncStarted = false;
    }

    ServletRequest req = event.getSuppliedRequest();
    asyncStartedCorrect = (expectedAsyncStarted == req.isAsyncStarted());
}
 

@Override
public void complete() {
	MockHttpServletRequest mockRequest = WebUtils.getNativeRequest(request, MockHttpServletRequest.class);
	if (mockRequest != null) {
		mockRequest.setAsyncStarted(false);
	}
	for (AsyncListener listener : this.listeners) {
		try {
			listener.onComplete(new AsyncEvent(this, this.request, this.response));
		}
		catch (IOException e) {
			throw new IllegalStateException("AsyncListener failure", e);
		}
	}
}
 

@Override
public void onError(AsyncEvent event) throws IOException {
    HttpServletRequest httpRequest = (HttpServletRequest) event.getSuppliedRequest();
    HttpServletResponse httpResponse = (HttpServletResponse) event.getSuppliedResponse();
    for (ServletFilterSpanDecorator spanDecorator: spanDecorators) {
        spanDecorator.onError(httpRequest,
            httpResponse,
            event.getThrowable(),
            span);
    }
}
 

@SuppressWarnings("unchecked")
@Test
public void onErrorHandlerAfterOnErrorEvent() throws Exception {
	Consumer<Throwable> handler = mock(Consumer.class);
	this.asyncRequest.addErrorHandler(handler);

	this.asyncRequest.startAsync();
	Exception e = new Exception();
	this.asyncRequest.onError(new AsyncEvent(this.request.getAsyncContext(), e));

	verify(handler).accept(e);
}
 

@Test
public void testServletUsesPassedInRecorder() throws IOException, ServletException {
    AWSXRayRecorder customRecorder = Mockito.spy(getMockRecorder());
    AWSXRayServletFilter servletFilter = new AWSXRayServletFilter(new FixedSegmentNamingStrategy("test"), customRecorder);

    AsyncContext asyncContext = Mockito.mock(AsyncContext.class);

    HttpServletRequest request = Mockito.mock(HttpServletRequest.class);
    Mockito.when(request.getRequestURL()).thenReturn(new StringBuffer("test_url"));
    Mockito.when(request.getMethod()).thenReturn("TEST_METHOD");
    Mockito.when(request.isAsyncStarted()).thenReturn(true);
    Mockito.when(request.getAsyncContext()).thenReturn(asyncContext);

    HttpServletResponse response = Mockito.mock(HttpServletResponse.class);

    FilterChain chain = mockChain(request, response);

    AsyncEvent event = Mockito.mock(AsyncEvent.class);
    Mockito.when(event.getSuppliedRequest()).thenReturn(request);
    Mockito.when(event.getSuppliedResponse()).thenReturn(response);

    servletFilter.doFilter(request, response, chain);
    Assert.assertNull(AWSXRay.getTraceEntity());

    AWSXRayServletAsyncListener listener = (AWSXRayServletAsyncListener) Whitebox.getInternalState(servletFilter, "listener");
    listener.onComplete(event);

    Mockito.verify(customRecorder.getEmitter(), Mockito.times(1)).sendSegment(Mockito.any());
}
 

private int getStatusCode(AsyncEvent asyncEvent) {
    try {
        if (asyncEvent.getSuppliedResponse() instanceof HttpServletResponse) {
            return ((HttpServletResponse) asyncEvent.getSuppliedResponse()).getStatus();
        }
    } catch (Exception ignored) {
    }
    return 0;
}
 

@Test
public void startAsyncAfterCompleted() throws Exception {
	this.asyncRequest.onComplete(new AsyncEvent(null));
	try {
		this.asyncRequest.startAsync();
		fail("expected exception");
	}
	catch (IllegalStateException ex) {
		assertEquals("Async processing has already completed", ex.getMessage());
	}
}
 

@Test
public void startDeferredResultProcessingErrorAndResumeWithDefaultResult() throws Exception {

	Exception e = new Exception();
	DeferredResult<Throwable> deferredResult = new DeferredResult<>(null, e);
	this.asyncManager.startDeferredResultProcessing(deferredResult);

	AsyncEvent event = new AsyncEvent(new MockAsyncContext(this.servletRequest, this.servletResponse), e);
	this.asyncWebRequest.onError(event);

	assertTrue(this.asyncManager.hasConcurrentResult());
	assertEquals(e, this.asyncManager.getConcurrentResult());
	assertEquals("/test", ((MockAsyncContext) this.servletRequest.getAsyncContext()).getDispatchedPath());
}
 

public void setStarted(Context context, ServletRequest request,
        ServletResponse response, boolean originalRequestResponse) {

    synchronized (asyncContextLock) {
        this.request.getCoyoteRequest().action(
                ActionCode.ASYNC_START, this);

        this.context = context;
        this.servletRequest = request;
        this.servletResponse = response;
        this.hasOriginalRequestAndResponse = originalRequestResponse;
        this.event = new AsyncEvent(this, request, response);

        List<AsyncListenerWrapper> listenersCopy =
            new ArrayList<AsyncListenerWrapper>();
        listenersCopy.addAll(listeners);
        listeners.clear();
        for (AsyncListenerWrapper listener : listenersCopy) {
            try {
                listener.fireOnStartAsync(event);
            } catch (Throwable t) {
                ExceptionUtils.handleThrowable(t);
                log.warn("onStartAsync() failed for listener of type [" +
                        listener.getClass().getName() + "]", t);
            }
        }
    }
}
 

@SuppressWarnings("unchecked")
@Test
public void onErrorHandlerAfterOnErrorEvent() throws Exception {
	Consumer<Throwable> handler = mock(Consumer.class);
	this.asyncRequest.addErrorHandler(handler);

	this.asyncRequest.startAsync();
	Exception e = new Exception();
	this.asyncRequest.onError(new AsyncEvent(this.request.getAsyncContext(), e));

	verify(handler).accept(e);
}
 

@Test
public void onCompletionHandlerAfterOnCompleteEvent() throws Exception {
	Runnable handler = mock(Runnable.class);
	this.asyncRequest.addCompletionHandler(handler);

	this.asyncRequest.startAsync();
	this.asyncRequest.onComplete(new AsyncEvent(this.request.getAsyncContext()));

	verify(handler).run();
	assertTrue(this.asyncRequest.isAsyncComplete());
}
 

@Override
public void complete() {
	MockHttpServletRequest mockRequest = WebUtils.getNativeRequest(this.request, MockHttpServletRequest.class);
	if (mockRequest != null) {
		mockRequest.setAsyncStarted(false);
	}
	for (AsyncListener listener : this.listeners) {
		try {
			listener.onComplete(new AsyncEvent(this, this.request, this.response));
		}
		catch (IOException ex) {
			throw new IllegalStateException("AsyncListener failure", ex);
		}
	}
}
 

@Override
public void onStartAsync(AsyncEvent asyncEvent) throws IOException {
}
 

@Override
public void onError(AsyncEvent event) throws IOException {
    event.getAsyncContext().complete();
}
 

@Override
public void onStartAsync(AsyncEvent event) throws IOException {
    TestAsyncContextImpl.track("onStartAsync-");
}