olympum · July 14, 2010 05:38
diff --git a/BigPipeContinuationServlet.java b/BigPipeContinuationServlet.java
 public class BigPipeContinuationServlet extends HttpServlet {

    private HttpClient httpClient;

    @Override
    public void init() throws ServletException {
        httpClient = new HttpClient();
        httpClient.setConnectorType(HttpClient.CONNECTOR_SELECT_CHANNEL);
        httpClient.setMaxConnectionsPerAddress(200); // max 200 concurrent connections to every address
        httpClient.setThreadPool(new QueuedThreadPool(250)); // max 250 threads
        httpClient.setTimeout(10000); // 10 seconds timeout; if no server reply, the request expires
        try {
            httpClient.start();
        } catch (Exception e) {
            throw new ServletException(e);
        }
    }

    @SuppressWarnings("unchecked")
    @Override
    protected void doGet(HttpServletRequest req, HttpServletResponse res) throws ServletException, IOException {
        final PrintWriter writer = res.getWriter();
        final Continuation continuation = ContinuationSupport.getContinuation(req);
        if (continuation.isInitial()) {
            ArrayList<String> pagelets = new ArrayList<String>();
            req.setAttribute("PAGELETS", pagelets);
            res.setStatus(200);
            res.setContentType("text/html");
            writer.write("<!DOCTYPE html PUBLIC \"-//W3C//DTD XHTML 1.0 Strict//EN\"" +
                    "   \"http://www.w3.org/TR/xhtml1/DTD/xhtml1-strict.dtd\">");
            writer.write("<html><head><title>bigpipe</title>");
            writer.write("</head><body><div><h1>Progressive Loading</h1>");
            for (int i = 0; i < 6; i++) {
                String pageletName = "pagelet" + i;
                pagelets.add(pageletName);
                writer.write("<div id='" + pageletName + "'></div>");
                // we are using the request attributes to track which pagelets have returned
                req.setAttribute(pageletName, pageletName);
            }
            writer.write("</div>");
            writer.write("<script type=\"text/javascript\">function arrived(id,text)" +
                    " { var b=document.getElementById(id); b.innerHTML = text; }</script>\n");
            // flush the buffer, otherwise the httpClient does not get anything
            writer.flush();

            continuation.suspend();

            for (String pagelet : pagelets) {
                final String pageletName = pagelet;
                ContentExchange exchange = new ContentExchange(true) {
                    @Override
                    protected void onResponseComplete() throws IOException {
                        // since the pagelet has returned, we can remove it
                        if (getResponseStatus() == 200) {
                            writer.write(composePagelet(pageletName, getResponseContent()));
                            // flush will cause the pagelet to appear in the browser
                            writer.flush();
                        }
                        // we can only remove the attribute after flushing, otherwise the check condition
                        // might return true and we would never flush
                        continuation.removeAttribute(pageletName);
                        if (continuation.isSuspended()) {
                            continuation.resume();
                        }
                    }

                    @Override
                    protected void onExpire() {
                        writer.write(composePagelet(pageletName, "some default content"));
                        writer.flush();
                    }
                };

                exchange.setMethod("GET");
                exchange.setURL("http://localhost:8080/module?id=" + pagelet);
                httpClient.send(exchange);
            }
        } else {
            // we can only check that all the pagelets returned from the servlet thread
            final ArrayList<String> pagelets = (ArrayList<String>) continuation.getAttribute("PAGELETS");
            if (pagelets == null) {
                return;
            }
            for (String pagelet : pagelets) {
                if (continuation.getAttribute(pagelet) != null) {
                    continuation.suspend();
                    return;
                }
            }
            continuation.resume();
            writer.write("</body></html>");
        }
    }

    private String composePagelet(String pageletName, String message) {
        return String.format("<script>arrived('%s', '%s');</script>", pageletName, message);
    }
 }
	public class BigPipeContinuationServlet extends HttpServlet {

	private HttpClient httpClient;

	@Override
	public void init() throws ServletException {
	httpClient = new HttpClient();
	httpClient.setConnectorType(HttpClient.CONNECTOR_SELECT_CHANNEL);
	httpClient.setMaxConnectionsPerAddress(200); // max 200 concurrent connections to every address
	httpClient.setThreadPool(new QueuedThreadPool(250)); // max 250 threads
	httpClient.setTimeout(10000); // 10 seconds timeout; if no server reply, the request expires
	try {
	httpClient.start();
	} catch (Exception e) {
	throw new ServletException(e);
	}
	}

	@SuppressWarnings("unchecked")
	@Override
	protected void doGet(HttpServletRequest req, HttpServletResponse res) throws ServletException, IOException {
	final PrintWriter writer = res.getWriter();
	final Continuation continuation = ContinuationSupport.getContinuation(req);
	if (continuation.isInitial()) {
	ArrayList<String> pagelets = new ArrayList<String>();
	req.setAttribute("PAGELETS", pagelets);
	res.setStatus(200);
	res.setContentType("text/html");
	writer.write("<!DOCTYPE html PUBLIC \"-//W3C//DTD XHTML 1.0 Strict//EN\"" +
	" \"http://www.w3.org/TR/xhtml1/DTD/xhtml1-strict.dtd\">");
	writer.write("<html><head><title>bigpipe</title>");
	writer.write("</head><body><div><h1>Progressive Loading</h1>");
	for (int i = 0; i < 6; i++) {
	String pageletName = "pagelet" + i;
	pagelets.add(pageletName);
	writer.write("<div id='" + pageletName + "'></div>");
	// we are using the request attributes to track which pagelets have returned
	req.setAttribute(pageletName, pageletName);
	}
	writer.write("</div>");
	writer.write("<script type=\"text/javascript\">function arrived(id,text)" +
	" { var b=document.getElementById(id); b.innerHTML = text; }</script>\n");
	// flush the buffer, otherwise the httpClient does not get anything
	writer.flush();

	continuation.suspend();

	for (String pagelet : pagelets) {
	final String pageletName = pagelet;
	ContentExchange exchange = new ContentExchange(true) {
	@Override
	protected void onResponseComplete() throws IOException {
	// since the pagelet has returned, we can remove it
	if (getResponseStatus() == 200) {
	writer.write(composePagelet(pageletName, getResponseContent()));
	// flush will cause the pagelet to appear in the browser
	writer.flush();
	}
	// we can only remove the attribute after flushing, otherwise the check condition
	// might return true and we would never flush
	continuation.removeAttribute(pageletName);
	if (continuation.isSuspended()) {
	continuation.resume();
	}
	}

	@Override
	protected void onExpire() {
	writer.write(composePagelet(pageletName, "some default content"));
	writer.flush();
	}
	};

	exchange.setMethod("GET");
	exchange.setURL("http://localhost:8080/module?id=" + pagelet);
	httpClient.send(exchange);
	}
	} else {
	// we can only check that all the pagelets returned from the servlet thread
	final ArrayList<String> pagelets = (ArrayList<String>) continuation.getAttribute("PAGELETS");
	if (pagelets == null) {
	return;
	}
	for (String pagelet : pagelets) {
	if (continuation.getAttribute(pagelet) != null) {
	continuation.suspend();
	return;
	}
	}
	continuation.resume();
	writer.write("</body></html>");
	}
	}

	private String composePagelet(String pageletName, String message) {
	return String.format("<script>arrived('%s', '%s');</script>", pageletName, message);
	}
	}