javax.servlet.ServletResponseWrapper#javax.servlet.AsyncContext源码实例Demo

@Override
protected void service(HttpServletRequest req, HttpServletResponse resp)
        throws ServletException, IOException {
    if (req.isAsyncStarted()) {
        req.getAsyncContext().complete();
    } else if (req.isAsyncSupported()) {
        AsyncContext actx = req.startAsync();
        actx.addListener(this);
        resp.setContentType("text/plain");
        clients.add(actx);
        if (clientcount.incrementAndGet()==1) {
            Stockticker ticker = (Stockticker) req.getServletContext().getAttribute(
                    AsyncStockContextListener.STOCK_TICKER_KEY);
            ticker.addTickListener(this);
        }
    } else {
        new Exception("Async Not Supported").printStackTrace();
        resp.sendError(400,"Async is not supported.");
    }
}
 

private boolean validate(final Object target, final Object result, final Throwable throwable) {
    if (throwable != null || result == null) {
        return false;
    }

    if (!(target instanceof HttpServletRequest)) {
        logger.debug("Invalid target object. {}", target);
        return false;
    }


    if (!(result instanceof AsyncContext)) {
        logger.debug("Invalid result object. {}.", result);
        return false;
    }

    return true;
}
 

public PortletAsyncContextImpl(AsyncContext actx, PortletResourceRequestContext prctx, ResourceRequest resreq, ResourceResponse resresp, boolean origReqResp) {
   this.actx = actx;
   this.prctx = prctx;
   this.resreq = resreq;
   this.resresp = resresp;
   this.hasOriginalReqResp = origReqResp;

   this.hreq = (HttpServletRequest) actx.getRequest();
   this.beanmgr = prctx.getBeanManager();

   // get the original container req & resp to pass to listener for resource releasing

   HttpServletRequest creq = prctx.getContainerRequest();
   HttpServletResponse cresp = prctx.getContainerResponse();

   pal = new PortletAsyncContextListener(this);
   actx.addListener(pal, creq, cresp);
}
 

private boolean validate(final Object target, final Object result, final Throwable throwable) {
    if (throwable != null || result == null) {
        return false;
    }

    if (!(target instanceof HttpServletRequest)) {
        if (isDebug) {
            logger.debug("Invalid target object, The javax.servlet.http.HttpServletRequest interface is not implemented. target={}", target);
        }
        return false;
    }

    if (!(result instanceof AsyncContext)) {
        if (isDebug) {
            logger.debug("Invalid result object, The javax.servlet.AsyncContext interface is not implemented. result={}.", result);
        }
        return false;
    }
    return true;
}
 

@Override
protected void doPut(HttpServletRequest req, final HttpServletResponse resp)
        throws ServletException, IOException {
    final AsyncContext ac = req.startAsync();
    ac.start(new Runnable() {
        @Override
        public void run() {
            resp.setContentType("text/plain");
            resp.setCharacterEncoding("UTF-8");
            try {
                resp.getWriter().print("OK");
            } catch (IOException e) {
                // Should never happen. Test will fail if it does.
            }
            ac.complete();
        }
    });
}
 

@Override
protected void service(final HttpServletRequest req, final HttpServletResponse resp) throws ServletException, IOException {
    final AsyncContext asyncCtx = req.startAsync();
    execService.execute(new Runnable() {
        @Override
        public void run() {
            try {
                req.login("test", "secret");
                resp.getWriter().println(user.getUser());
            } catch (final Exception e) {
                try {
                    e.printStackTrace(resp.getWriter());
                } catch (final IOException e1) {
                    throw new IllegalStateException(e);
                }
            } finally {
                asyncCtx.complete();
            }
        }
    });
}
 

@Override
protected void service(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {

	final AsyncContext asyncContext = req.startAsync();

	// DON'T TRY THIS AT HOME: Working with threads in a Container !!
	new Timer().schedule(new TimerTask() {
		@Override
		public void run() {
			try {
				asyncContext.getResponse().getWriter().println(new Date() + ": Asynchronous Response");
				System.out.println("Release pending HTTP connection");
				asyncContext.complete();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}, 10 * 1000);
	resp.getWriter().println(new Date() + ": Async Servlet returns leaving the connection open");
}
 

@Override
protected void doGet(final HttpServletRequest req,
        final HttpServletResponse resp)
        throws ServletException, IOException {

    final AsyncContext async = req.startAsync();
    // Just for debugging
    async.setTimeout(100000);

    ExecutorService executor = Executors.newSingleThreadExecutor();
    executor.submit(new Runnable() {

        @Override
        public void run() {
            async.dispatch("/ServletC");
        }
    });
    executor.shutdown();
}
 

@Override
protected void doGet(HttpServletRequest req, HttpServletResponse resp)
        throws ServletException, IOException {

    resp.setContentType("text/plain");
    resp.setCharacterEncoding("UTF-8");
    ServletOutputStream sos = resp.getOutputStream();


    AsyncContext asyncCtxt = req.startAsync();
    asyncCtxt.setTimeout(5);
    Runnable task = new AsyncTask(asyncCtxt, sos);
    if (useContainerThreadToSetListener) {
        asyncCtxt.start(task);
    } else {
        Thread t = new Thread(task);
        t.start();
    }
}
 

@Override
public void onExitStartAsync(AsyncContext asyncContext) {
    final ServletRequest request = asyncContext.getRequest();
    if (request.getAttribute(ASYNC_LISTENER_ADDED) != null) {
        return;
    }
    final Transaction transaction = tracer.currentTransaction();
    if (transaction != null && transaction.isSampled() && request.getAttribute(ASYNC_LISTENER_ADDED) == null) {
        // makes sure that the listener is only added once, even if the request is wrapped
        // which leads to multiple invocations of startAsync for the same underlying request
        request.setAttribute(ASYNC_LISTENER_ADDED, Boolean.TRUE);
        // specifying the request and response is important
        // otherwise AsyncEvent.getSuppliedRequest returns null per spec
        // however, only some application server like WebSphere actually implement it that way
        asyncContext.addListener(asyncListenerObjectPool.createInstance().withTransaction(transaction),
            asyncContext.getRequest(), asyncContext.getResponse());

        request.setAttribute(ASYNC_ATTRIBUTE, Boolean.TRUE);
        request.setAttribute(TRANSACTION_ATTRIBUTE, transaction);
    }
}
 

@Override
public void doGet(final HttpServletRequest request, final HttpServletResponse response)
    throws ServletException, IOException {

    // avoid retries on timeout
    if (request.getAttribute("timedOut") != null) {
        return;
    }
    request.setAttribute("timedOut", true);

    final AsyncContext asyncContext = request.startAsync(request, response);
    asyncContext.setTimeout(10);
    asyncContext.start(new Runnable() {
        @Override
        public void run() {
            try {
                TimeUnit.MILLISECONDS.sleep(200);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                asyncContext.complete();
            }
        }
    });
}
 

@Override
protected void doGet(HttpServletRequest req, final HttpServletResponse resp)
        throws ServletException, IOException {

    final AsyncContext actxt = req.startAsync();
    actxt.setTimeout(TIMEOUT);
    if (threaded) {
        actxt.start(new Runnable() {
            @Override
            public void run() {
                try {
                    resp.sendError(status, ERROR_MESSAGE);
                    actxt.complete();
                } catch (IOException e) {
                    // Ignore
                }
            }
        });
    } else {
        resp.sendError(status, ERROR_MESSAGE);
        actxt.complete();
    }
}
 

public ServletServerHttpResponse(HttpHeaders headers, HttpServletResponse response, AsyncContext asyncContext,
		DataBufferFactory bufferFactory, int bufferSize, ServletServerHttpRequest request) throws IOException {

	super(bufferFactory, headers);

	Assert.notNull(response, "HttpServletResponse must not be null");
	Assert.notNull(bufferFactory, "DataBufferFactory must not be null");
	Assert.isTrue(bufferSize > 0, "Buffer size must be greater than 0");

	this.response = response;
	this.outputStream = response.getOutputStream();
	this.bufferSize = bufferSize;
	this.request = request;

	asyncContext.addListener(new ResponseAsyncListener());

	// Tomcat expects WriteListener registration on initial thread
	response.getOutputStream().setWriteListener(new ResponseBodyWriteListener());
}
 

@Override
protected void doGet(HttpServletRequest req, final HttpServletResponse resp)
        throws ServletException, IOException {

    final AsyncContext actxt = req.startAsync();
    actxt.setTimeout(TIMEOUT);
    if (threaded) {
        actxt.start(new Runnable() {
            @Override
            public void run() {
                try {
                    resp.sendError(status, ERROR_MESSAGE);
                    actxt.complete();
                } catch (IOException e) {
                    // Ignore
                }
            }
        });
    } else {
        resp.sendError(status, ERROR_MESSAGE);
        actxt.complete();
    }
}
 

public MainHandler(LinkServer link) {
    this.link = link;
    link.onResponse((ResponseMessage res) -> {
        AsyncContext ctx = contexts.get(res.getId());
        if (ctx == null) {
            System.out.println("ERROR: Could not find context for " + res.getId());
            return;
        }

        HttpServletResponse resp = (HttpServletResponse)ctx.getResponse();
        if (resp == null) {
            System.out.println("ERROR: Could not find response object for " + res.getId());
            return;
        }

        try {
            resp.setStatus(res.getStatus() != 0 ? res.getStatus() : HttpServletResponse.SC_OK);
            res.getHeaders().forEach(resp::setHeader);
            resp.getWriter().write(res.getMessage());
            ctx.complete();
        } catch (IOException e) {
            e.printStackTrace();
        }
    });
}
 

public void service(HttpServletRequest request, HttpServletResponse response) {
  if (transport == null) {
    transport = SCBEngine.getInstance().getTransportManager().findTransport(Const.RESTFUL);
    microserviceMeta = SCBEngine.getInstance().getProducerMicroserviceMeta();
  }

  // 异步场景
  AsyncContext asyncCtx = request.startAsync();
  asyncCtx.addListener(restAsyncListener);
  asyncCtx.setTimeout(ServletConfig.getAsyncServletTimeout());

  HttpServletRequestEx requestEx = new StandardHttpServletRequestEx(request);
  HttpServletResponseEx responseEx = new StandardHttpServletResponseEx(response);

  if (SCBEngine.getInstance().isFilterChainEnabled()) {
    ((StandardHttpServletRequestEx) requestEx).setCacheRequest(true);
    InvocationCreator creator = new RestServletProducerInvocationCreator(microserviceMeta, transport.getEndpoint(),
        requestEx, responseEx);
    new RestProducerInvocationFlow(creator, requestEx, responseEx)
        .run();
    return;
  }

  RestServletProducerInvocation restProducerInvocation = new RestServletProducerInvocation();
  restProducerInvocation.invoke(transport, requestEx, responseEx, httpServerFilters);
}
 

public void writeStock(AsyncContext actx, Stock stock) {
    HttpServletResponse response = (HttpServletResponse)actx.getResponse();
    try {
        PrintWriter writer = response.getWriter();
        writer.write("STOCK#");//make client parsing easier
        writer.write(stock.getSymbol());
        writer.write("#");
        writer.write(stock.getValueAsString());
        writer.write("#");
        writer.write(stock.getLastChangeAsString());
        writer.write("#");
        writer.write(String.valueOf(stock.getCnt()));
        writer.write("\n");
        writer.flush();
        response.flushBuffer();
    }catch (IOException x) {
        try {actx.complete();}catch (Exception ignore){/* Ignore */}
    }
}
 

/**
 * 异步调用
 * 
 * @param message message
 * @param ctx {@link AsyncContext}
 */
public <T> void asyncCallService(T message, AsyncContext ctx) {
  ServiceContext serviceContext = null;
  if (ctx != null) {
    Web web = new Web(ctx);
    serviceContext = ServiceContextUtil.context(message, web);
    ServiceContextUtil.record(serviceContext);
  } else {
    serviceContext = ServiceContextUtil.context(message);
  }
  serviceContext.setFlowName(headerServiceConfig.getFlowName());
  if (StringUtil.isBlank(serviceContext.getCurrentServiceName())) {
    serviceContext.setCurrentServiceName(headerServiceConfig.getServiceName());
  }
  serviceContext.addAttachment(Constant.ServiceContextOriginURL,
      headerServiceConfig.getAddresses().iterator().next());
  getServiceRouter().asyncCallService(serviceContext, ActorRef.noSender());
}
 

private NumberWriterListener(AsyncContext ac, ServletInputStream sis,
        ServletOutputStream sos) {
    this.ac = ac;
    this.sis = sis;
    this.sos = sos;

    // In Tomcat, the order the listeners are set controls the order
    // that the first calls are made. In this case, the read listener
    // will be called before the write listener.
    sis.setReadListener(this);
    sos.setWriteListener(this);
}
 

@Override
public void tick(Stock stock) {
    Iterator<AsyncContext> it = clients.iterator();
    while (it.hasNext()) {
        AsyncContext actx = it.next();
        try {
            writeStock(actx, stock);
        } catch (Exception e) {
            // Ignore. The async error handling will deal with this.
        }
    }
}
 

@Override
public void shutdown() {
    // The web application is shutting down. Complete any AsyncContexts
    // associated with an active client.
    Iterator<AsyncContext> it = clients.iterator();
    while (it.hasNext()) {
        AsyncContext actx = it.next();
        try {
            actx.complete();
        } catch (Exception e) {
            // Ignore. The async error handling will deal with this.
        }
    }
}
 

@Override
public AsyncContext getAsyncContext() {
    if (!isAsyncStarted()) {
        throw new IllegalStateException(sm.getString("request.notAsync"));
    }
    return asyncContext;
}
 

@Override
public void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
	AsyncContext context = request.startAsync();
	synchronized (asyncContexts) {
		asyncContexts.add(context);
	}
}
 

@Override
public AsyncContext startAsync() {
	// issue 217: after MonitoringFilter.doFilter, response is instance of CounterServletResponseWrapper,
	// and if response.getWriter() has been called before calling request.startAsync(),
	// then asyncContext.getResponse() should return the instance of CounterServletResponseWrapper
	// and not the initial response without the wrapper,
	// otherwise asyncContext.getResponse().getWriter() will throw something like
	// "IllegalStateException: getOutputStream() has already been called for this response"
	return super.startAsync(this, response);
}
 

public Web(AsyncContext context) {
  this.asyncContext = context;
  this.servletRequest = context.getRequest();
  this.servletResponse = context.getResponse();
  servletResponse.setCharacterEncoding(Constant.ENCODING_UTF_8);
  try {
    this.writer = servletResponse.getWriter();
  } catch (IOException e) {
    logger.error("", e);
  }
}
 

@Override
protected void doGet(HttpServletRequest req, HttpServletResponse resp)
        throws ServletException, IOException {

    // Should not be async at this point
    isAsyncWhenExpected = isAsyncWhenExpected && !req.isAsyncStarted();

    final AsyncContext async = req.startAsync();

    // Should be async at this point
    isAsyncWhenExpected = isAsyncWhenExpected && req.isAsyncStarted();

    async.start(new Runnable() {

        @Override
        public void run() {
            // This should be delayed until the original container
            // thread exists
            async.dispatch("/ServletB");
        }
    });

    try {
        Thread.sleep(3000);
    } catch (InterruptedException e) {
        throw new ServletException(e);
    }

    // Should be async at this point
    isAsyncWhenExpected = isAsyncWhenExpected && req.isAsyncStarted();
}
 

@Override
public AsyncContext startAsync(
        final ServletRequest wrappedRequest,
        final ServletResponse wrappedResponse) {
    // According to the documentation, the passed request/response parameters here must either
    // _be_ or _wrap_ the original request/response objects passed to the servlet - which are
    // our wrappers, so no need to wrap again - we can use the user-supplied objects.
    final AsyncContext asyncContext = super.startAsync(wrappedRequest, wrappedResponse);
    return new FilterInvokingAsyncContext(asyncContext, filterInvoker, this, wrappedResponse);
}
 

DatabusPollRunnable(AsyncContext asyncContext, KeepAliveRunnable keepAliveRunnable, Subject subject, SubjectDatabus databus,
                    Duration claimTtl, int limit, String subscription, PeekOrPollResponseHelper helper,
                    long longPollStopTime, Timer.Context timerContext) {
    _asyncContext = asyncContext;
    _keepAliveRunnable = keepAliveRunnable;
    _subject = subject;
    _databus = databus;
    _claimTtl = claimTtl;
    _limit = limit;
    _subscription = subscription;
    _helper = helper;
    _longPollStopTime = longPollStopTime;
    _timerContext = timerContext;
}
 

public void doGet(HttpServletRequest request, HttpServletResponse response) {
    final AsyncContext asyncContext = request.startAsync(request, response);

    executor.execute(() -> {
        sleepThread(SLEEP_TIME_MILLIS);
        asyncContext.dispatch(ASYNC_PATH);
    });
}
 

private void asyncExe(AsyncContext ctx) {
  try {
    sr.asyncCallService(null, ctx);
  } catch (Exception e) {
    e.printStackTrace();
  }
}