////  ========================================================================//  Copyright (c) 1995-2016 Mort Bay Consulting Pty. Ltd.//  ------------------------------------------------------------------------//  All rights reserved. This program and the accompanying materials//  are made available under the terms of the Eclipse Public License v1.0//  and Apache License v2.0 which accompanies this distribution.////      The Eclipse Public License is available at//      http://www.eclipse.org/legal/epl-v10.html////      The Apache License v2.0 is available at//      http://www.opensource.org/licenses/apache2.0.php////  You may elect to redistribute this code under either of these licenses.//  ========================================================================//package com.acme;import java.io.IOException;import java.io.PrintWriter;import java.util.HashMap;import java.util.LinkedList;import java.util.Map;import java.util.Queue;import javax.servlet.ServletException;import javax.servlet.http.HttpServlet;import javax.servlet.http.HttpServletRequest;import javax.servlet.http.HttpServletResponse;import org.eclipse.jetty.continuation.Continuation;import org.eclipse.jetty.continuation.ContinuationSupport;// 简单的异步聊天室// 这不处理重复的用户名和同一个浏览器的标签情况// 一些代码是重复的public class ChatServlet extends HttpServlet{        // 内部类用来保存每个成员的消息队列    class Member    {        String _name;        Continuation _continuation;        Queue<String> _queue = new LinkedList<String>();    }    Map<String,Map<String,Member>> _rooms = new HashMap<String,Map<String, Member>>();            // 处理浏览器的AJAX调用    @Override    protected void doPost(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException    {           // Ajax调用的编码形式        String action = request.getParameter("action");        String message = request.getParameter("message");        String username = request.getParameter("user");        if (action.equals("join"))            join(request,response,username);        else if (action.equals("poll"))            poll(request,response,username);        else if (action.equals("chat"))            chat(request,response,username,message);    }    private synchronized void join(HttpServletRequest request,HttpServletResponse response,String username)    throws IOException    {        Member member = new Member();        member._name=username;        Map<String,Member> room=_rooms.get(request.getPathInfo());        if (room==null)        {            room=new HashMap<String,Member>();            _rooms.put(request.getPathInfo(),room);        }        room.put(username,member);         response.setContentType("text/json;charset=utf-8");        PrintWriter out=response.getWriter();        out.print("{action:\"join\"}");    }    private synchronized void poll(HttpServletRequest request,HttpServletResponse response,String username)    throws IOException    {        Map<String,Member> room=_rooms.get(request.getPathInfo());        if (room==null)        {            response.sendError(503);            return;        }        Member member = room.get(username);        if (member==null)        {            response.sendError(503);            return;        }        synchronized(member)        {            if (member._queue.size()>0)            {                // 发送一个聊天消息                response.setContentType("text/json;charset=utf-8");                StringBuilder buf=new StringBuilder();                buf.append("{\"action\":\"poll\",");                buf.append("\"from\":\"");                buf.append(member._queue.poll());                buf.append("\",");                String message = member._queue.poll();                int quote=message.indexOf(‘"‘);                while (quote>=0)                {                    message=message.substring(0,quote)+‘\\‘+message.substring(quote);                    quote=message.indexOf(‘"‘,quote+2);                }                buf.append("\"chat\":\"");                buf.append(message);                buf.append("\"}");                byte[] bytes = buf.toString().getBytes("utf-8");                response.setContentLength(bytes.length);                response.getOutputStream().write(bytes);            }            else             {                Continuation continuation = ContinuationSupport.getContinuation(request);                if (continuation.isInitial())                 {                    // 没有消息时，挂起等待聊天或超时                    continuation.setTimeout(20000);                    continuation.suspend();                    member._continuation=continuation;                }                else                {                    // 超时后发送空的响应                    response.setContentType("text/json;charset=utf-8");                    PrintWriter out=response.getWriter();                    out.print("{action:\"poll\"}");                }            }        }    }    private synchronized void chat(HttpServletRequest request,HttpServletResponse response,String username,String message)    throws IOException    {        Map<String,Member> room=_rooms.get(request.getPathInfo());        if (room!=null)        {            // 推送消息到所有成员            for (Member m:room.values())            {                synchronized (m)                {                    m._queue.add(username); // from                    m._queue.add(message);  // chat                    // 如果轮询到则唤醒                    if (m._continuation!=null)                    {                        m._continuation.resume();                        m._continuation=null;                    }                }            }        }        response.setContentType("text/json;charset=utf-8");        PrintWriter out=response.getWriter();        out.print("{action:\"chat\"}");      }        // 提供嵌入css和js的html服务    // 这应该是静态内容和真正使用的js库    @Override    protected void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException    {        if (request.getParameter("action")!=null)            doPost(request,response);        else            getServletContext().getNamedDispatcher("default").forward(request,response);    }    }

////  ========================================================================//  Copyright (c) 1995-2016 Mort Bay Consulting Pty. Ltd.//  ------------------------------------------------------------------------//  All rights reserved. This program and the accompanying materials//  are made available under the terms of the Eclipse Public License v1.0//  and Apache License v2.0 which accompanies this distribution.////      The Eclipse Public License is available at//      http://www.eclipse.org/legal/epl-v10.html////      The Apache License v2.0 is available at//      http://www.opensource.org/licenses/apache2.0.php////  You may elect to redistribute this code under either of these licenses.//  ========================================================================//package org.eclipse.jetty.servlets;import java.io.IOException;import java.util.Queue;import java.util.concurrent.ConcurrentLinkedQueue;import java.util.concurrent.Semaphore;import java.util.concurrent.TimeUnit;import javax.servlet.AsyncContext;import javax.servlet.AsyncEvent;import javax.servlet.AsyncListener;import javax.servlet.Filter;import javax.servlet.FilterChain;import javax.servlet.FilterConfig;import javax.servlet.ServletContext;import javax.servlet.ServletException;import javax.servlet.ServletRequest;import javax.servlet.ServletResponse;import javax.servlet.http.HttpServletRequest;import javax.servlet.http.HttpServletResponse;import javax.servlet.http.HttpSession;import org.eclipse.jetty.server.handler.ContextHandler;import org.eclipse.jetty.util.annotation.ManagedAttribute;import org.eclipse.jetty.util.annotation.ManagedObject;import org.eclipse.jetty.util.log.Log;import org.eclipse.jetty.util.log.Logger;/** * Quality of Service Filter. * <p> * This filter limits the number of active requests to the number set by the "maxRequests" init parameter (default 10). * If more requests are received, they are suspended and placed on priority queues.  Priorities are determined by * the {@link #getPriority(ServletRequest)} method and are a value between 0 and the value given by the "maxPriority" * init parameter (default 10), with higher values having higher priority. * <p> * This filter is ideal to prevent wasting threads waiting for slow/limited * resources such as a JDBC connection pool.  It avoids the situation where all of a * containers thread pool may be consumed blocking on such a slow resource. * By limiting the number of active threads, a smaller thread pool may be used as * the threads are not wasted waiting.  Thus more memory may be available for use by * the active threads. * <p> * Furthermore, this filter uses a priority when resuming waiting requests. So that if * a container is under load, and there are many requests waiting for resources, * the {@link #getPriority(ServletRequest)} method is used, so that more important * requests are serviced first.     For example, this filter could be deployed with a * maxRequest limit slightly smaller than the containers thread pool and a high priority * allocated to admin users.  Thus regardless of load, admin users would always be * able to access the web application. * <p> * The maxRequest limit is policed by a {@link Semaphore} and the filter will wait a short while attempting to acquire * the semaphore. This wait is controlled by the "waitMs" init parameter and allows the expense of a suspend to be * avoided if the semaphore is shortly available.  If the semaphore cannot be obtained, the request will be suspended * for the default suspend period of the container or the valued set as the "suspendMs" init parameter. * <p> * If the "managedAttr" init parameter is set to true, then this servlet is set as a {@link ServletContext} attribute with the * filter name as the attribute name.  This allows context external mechanism (eg JMX via {@link ContextHandler#MANAGED_ATTRIBUTES}) to * manage the configuration of the filter. */@ManagedObject("Quality of Service Filter")public class QoSFilter implements Filter{    private static final Logger LOG = Log.getLogger(QoSFilter.class);    static final int __DEFAULT_MAX_PRIORITY = 10;    static final int __DEFAULT_PASSES = 10;    static final int __DEFAULT_WAIT_MS = 50;    static final long __DEFAULT_TIMEOUT_MS = -1;    static final String MANAGED_ATTR_INIT_PARAM = "managedAttr";    static final String MAX_REQUESTS_INIT_PARAM = "maxRequests";    static final String MAX_PRIORITY_INIT_PARAM = "maxPriority";    static final String MAX_WAIT_INIT_PARAM = "waitMs";    static final String SUSPEND_INIT_PARAM = "suspendMs";    private final String _suspended = "QoSFilter@" + Integer.toHexString(hashCode()) + ".SUSPENDED";    private final String _resumed = "QoSFilter@" + Integer.toHexString(hashCode()) + ".RESUMED";    private long _waitMs;    private long _suspendMs;    private int _maxRequests;    private Semaphore _passes;    private Queue<AsyncContext>[] _queues;    private AsyncListener[] _listeners;    public void init(FilterConfig filterConfig)    {        int max_priority = __DEFAULT_MAX_PRIORITY;        if (filterConfig.getInitParameter(MAX_PRIORITY_INIT_PARAM) != null)            max_priority = Integer.parseInt(filterConfig.getInitParameter(MAX_PRIORITY_INIT_PARAM));        _queues = new Queue[max_priority + 1];        _listeners = new AsyncListener[_queues.length];        for (int p = 0; p < _queues.length; ++p)        {            _queues[p] = new ConcurrentLinkedQueue<>();            _listeners[p] = new QoSAsyncListener(p);        }        int maxRequests = __DEFAULT_PASSES;        if (filterConfig.getInitParameter(MAX_REQUESTS_INIT_PARAM) != null)            maxRequests = Integer.parseInt(filterConfig.getInitParameter(MAX_REQUESTS_INIT_PARAM));        _passes = new Semaphore(maxRequests, true);        _maxRequests = maxRequests;        long wait = __DEFAULT_WAIT_MS;        if (filterConfig.getInitParameter(MAX_WAIT_INIT_PARAM) != null)            wait = Integer.parseInt(filterConfig.getInitParameter(MAX_WAIT_INIT_PARAM));        _waitMs = wait;        long suspend = __DEFAULT_TIMEOUT_MS;        if (filterConfig.getInitParameter(SUSPEND_INIT_PARAM) != null)            suspend = Integer.parseInt(filterConfig.getInitParameter(SUSPEND_INIT_PARAM));        _suspendMs = suspend;        ServletContext context = filterConfig.getServletContext();        if (context != null && Boolean.parseBoolean(filterConfig.getInitParameter(MANAGED_ATTR_INIT_PARAM)))            context.setAttribute(filterConfig.getFilterName(), this);    }    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException    {        boolean accepted = false;        try        {            Boolean suspended = (Boolean)request.getAttribute(_suspended);            if (suspended == null)            {                accepted = _passes.tryAcquire(getWaitMs(), TimeUnit.MILLISECONDS);                if (accepted)                {                    request.setAttribute(_suspended, Boolean.FALSE);                    if (LOG.isDebugEnabled())                        LOG.debug("Accepted {}", request);                }                else                {                    request.setAttribute(_suspended, Boolean.TRUE);                    int priority = getPriority(request);                    AsyncContext asyncContext = request.startAsync();                    long suspendMs = getSuspendMs();                    if (suspendMs > 0)                        asyncContext.setTimeout(suspendMs);                    asyncContext.addListener(_listeners[priority]);                    _queues[priority].add(asyncContext);                    if (LOG.isDebugEnabled())                        LOG.debug("Suspended {}", request);                    return;                }            }            else            {                if (suspended)                {                    request.setAttribute(_suspended, Boolean.FALSE);                    Boolean resumed = (Boolean)request.getAttribute(_resumed);                    if (resumed == Boolean.TRUE)                    {                        _passes.acquire();                        accepted = true;                        if (LOG.isDebugEnabled())                            LOG.debug("Resumed {}", request);                    }                    else                    {                        // Timeout! try 1 more time.                        accepted = _passes.tryAcquire(getWaitMs(), TimeUnit.MILLISECONDS);                        if (LOG.isDebugEnabled())                            LOG.debug("Timeout {}", request);                    }                }                else                {                    // Pass through resume of previously accepted request.                    _passes.acquire();                    accepted = true;                    if (LOG.isDebugEnabled())                        LOG.debug("Passthrough {}", request);                }            }            if (accepted)            {                chain.doFilter(request, response);            }            else            {                if (LOG.isDebugEnabled())                    LOG.debug("Rejected {}", request);                ((HttpServletResponse)response).sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE);            }        }        catch (InterruptedException e)        {            ((HttpServletResponse)response).sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE);        }        finally        {            if (accepted)            {                for (int p = _queues.length - 1; p >= 0; --p)                {                    AsyncContext asyncContext = _queues[p].poll();                    if (asyncContext != null)                    {                        ServletRequest candidate = asyncContext.getRequest();                        Boolean suspended = (Boolean)candidate.getAttribute(_suspended);                        if (suspended == Boolean.TRUE)                        {                            candidate.setAttribute(_resumed, Boolean.TRUE);                            asyncContext.dispatch();                            break;                        }                    }                }                _passes.release();            }        }    }    /**     * Computes the request priority.     * <p>     * The default implementation assigns the following priorities:     * <ul>     * <li> 2 - for an authenticated request     * <li> 1 - for a request with valid / non new session     * <li> 0 - for all other requests.     * </ul>     * This method may be overridden to provide application specific priorities.     *     * @param request the incoming request     * @return the computed request priority     */    protected int getPriority(ServletRequest request)    {        HttpServletRequest baseRequest = (HttpServletRequest)request;        if (baseRequest.getUserPrincipal() != null)        {            return 2;        }        else        {            HttpSession session = baseRequest.getSession(false);            if (session != null && !session.isNew())                return 1;            else                return 0;        }    }    public void destroy()    {    }    /**     * Get the (short) amount of time (in milliseconds) that the filter would wait     * for the semaphore to become available before suspending a request.     *     * @return wait time (in milliseconds)     */    @ManagedAttribute("(short) amount of time filter will wait before suspending request (in ms)")    public long getWaitMs()    {        return _waitMs;    }    /**     * Set the (short) amount of time (in milliseconds) that the filter would wait     * for the semaphore to become available before suspending a request.     *     * @param value wait time (in milliseconds)     */    public void setWaitMs(long value)    {        _waitMs = value;    }    /**     * Get the amount of time (in milliseconds) that the filter would suspend     * a request for while waiting for the semaphore to become available.     *     * @return suspend time (in milliseconds)     */    @ManagedAttribute("amount of time filter will suspend a request for while waiting for the semaphore to become available (in ms)")    public long getSuspendMs()    {        return _suspendMs;    }    /**     * Set the amount of time (in milliseconds) that the filter would suspend     * a request for while waiting for the semaphore to become available.     *     * @param value suspend time (in milliseconds)     */    public void setSuspendMs(long value)    {        _suspendMs = value;    }    /**     * Get the maximum number of requests allowed to be processed     * at the same time.     *     * @return maximum number of requests     */    @ManagedAttribute("maximum number of requests to allow processing of at the same time")    public int getMaxRequests()    {        return _maxRequests;    }    /**     * Set the maximum number of requests allowed to be processed     * at the same time.     *     * @param value the number of requests     */    public void setMaxRequests(int value)    {        _passes = new Semaphore((value - getMaxRequests() + _passes.availablePermits()), true);        _maxRequests = value;    }    private class QoSAsyncListener implements AsyncListener    {        private final int priority;        public QoSAsyncListener(int priority)        {            this.priority = priority;        }        @Override        public void onStartAsync(AsyncEvent event) throws IOException        {        }        @Override        public void onComplete(AsyncEvent event) throws IOException        {        }        @Override        public void onTimeout(AsyncEvent event) throws IOException        {            // Remove before it‘s redispatched, so it won‘t be            // redispatched again at the end of the filtering.            AsyncContext asyncContext = event.getAsyncContext();            _queues[priority].remove(asyncContext);            asyncContext.dispatch();        }        @Override        public void one rror(AsyncEvent event) throws IOException        {        }    }}

////  ========================================================================//  Copyright (c) 1995-2016 Mort Bay Consulting Pty. Ltd.//  ------------------------------------------------------------------------//  All rights reserved. This program and the accompanying materials//  are made available under the terms of the Eclipse Public License v1.0//  and Apache License v2.0 which accompanies this distribution.////      The Eclipse Public License is available at//      http://www.eclipse.org/legal/epl-v10.html////      The Apache License v2.0 is available at//      http://www.opensource.org/licenses/apache2.0.php////  You may elect to redistribute this code under either of these licenses.//  ========================================================================//package org.eclipse.jetty.proxy;import java.io.IOException;import java.io.InputStream;import java.nio.ByteBuffer;import java.util.concurrent.TimeUnit;import javax.servlet.AsyncContext;import javax.servlet.ServletConfig;import javax.servlet.ServletException;import javax.servlet.http.HttpServletRequest;import javax.servlet.http.HttpServletResponse;import org.eclipse.jetty.client.api.ContentProvider;import org.eclipse.jetty.client.api.Request;import org.eclipse.jetty.client.api.Response;import org.eclipse.jetty.client.api.Result;import org.eclipse.jetty.client.util.InputStreamContentProvider;import org.eclipse.jetty.http.HttpVersion;import org.eclipse.jetty.util.Callback;/** * <p>Servlet 3.0 asynchronous proxy servlet.</p> * <p>The request processing is asynchronous, but the I/O is blocking.</p> * * @see AsyncProxyServlet * @see AsyncMiddleManServlet * @see ConnectHandler */public class ProxyServlet extends AbstractProxyServlet{    @Override    protected void service(final HttpServletRequest request, final HttpServletResponse response) throws ServletException, IOException    {        final int requestId = getRequestId(request);        String rewrittenTarget = rewriteTarget(request);        if (_log.isDebugEnabled())        {            StringBuffer uri = request.getRequestURL();            if (request.getQueryString() != null)                uri.append("?").append(request.getQueryString());            if (_log.isDebugEnabled())                _log.debug("{} rewriting: {} -> {}", requestId, uri, rewrittenTarget);        }        if (rewrittenTarget == null)        {            onProxyRewriteFailed(request, response);            return;        }        final Request proxyRequest = getHttpClient().newRequest(rewrittenTarget)                .method(request.getMethod())                .version(HttpVersion.fromString(request.getProtocol()));        copyRequestHeaders(request, proxyRequest);        addProxyHeaders(request, proxyRequest);        final AsyncContext asyncContext = request.startAsync();        // We do not timeout the continuation, but the proxy request        asyncContext.setTimeout(0);        proxyRequest.timeout(getTimeout(), TimeUnit.MILLISECONDS);        if (hasContent(request))            proxyRequest.content(proxyRequestContent(request, response, proxyRequest));        sendProxyRequest(request, response, proxyRequest);    }    protected ContentProvider proxyRequestContent(HttpServletRequest request, HttpServletResponse response, Request proxyRequest) throws IOException    {        return new ProxyInputStreamContentProvider(request, response, proxyRequest, request.getInputStream());    }    @Override    protected Response.Listener newProxyResponseListener(HttpServletRequest request, HttpServletResponse response)    {        return new ProxyResponseListener(request, response);    }    protected void onResponseContent(HttpServletRequest request, HttpServletResponse response, Response proxyResponse, byte[] buffer, int offset, int length, Callback callback)    {        try        {            if (_log.isDebugEnabled())                _log.debug("{} proxying content to downstream: {} bytes", getRequestId(request), length);            response.getOutputStream().write(buffer, offset, length);            callback.succeeded();        }        catch (Throwable x)        {            callback.failed(x);        }    }    /**     * <p>Convenience extension of {@link ProxyServlet} that offers transparent proxy functionalities.</p>     *     * @see org.eclipse.jetty.proxy.AbstractProxyServlet.TransparentDelegate     */    public static class Transparent extends ProxyServlet    {        private final TransparentDelegate delegate = new TransparentDelegate(this);        @Override        public void init(ServletConfig config) throws ServletException        {            super.init(config);            delegate.init(config);        }        @Override        protected String rewriteTarget(HttpServletRequest request)        {            return delegate.rewriteTarget(request);        }    }    protected class ProxyResponseListener extends Response.Listener.Adapter    {        private final HttpServletRequest request;        private final HttpServletResponse response;        protected ProxyResponseListener(HttpServletRequest request, HttpServletResponse response)        {            this.request = request;            this.response = response;        }        @Override        public void onBegin(Response proxyResponse)        {            response.setStatus(proxyResponse.getStatus());        }        @Override        public void onHeaders(Response proxyResponse)        {            onServerResponseHeaders(request, response, proxyResponse);        }        @Override        public void onContent(final Response proxyResponse, ByteBuffer content, final Callback callback)        {            byte[] buffer;            int offset;            int length = content.remaining();            if (content.hasArray())            {                buffer = content.array();                offset = content.arrayOffset();            }            else            {                buffer = new byte[length];                content.get(buffer);                offset = 0;            }            onResponseContent(request, response, proxyResponse, buffer, offset, length, new Callback.Nested(callback)            {                @Override                public void failed(Throwable x)                {                    super.failed(x);                    proxyResponse.abort(x);                }            });        }        @Override        public void onComplete(Result result)        {            if (result.isSucceeded())                onProxyResponseSuccess(request, response, result.getResponse());            else                onProxyResponseFailure(request, response, result.getResponse(), result.getFailure());            if (_log.isDebugEnabled())                _log.debug("{} proxying complete", getRequestId(request));        }    }    protected class ProxyInputStreamContentProvider extends InputStreamContentProvider    {        private final HttpServletResponse response;        private final Request proxyRequest;        private final HttpServletRequest request;        protected ProxyInputStreamContentProvider(HttpServletRequest request, HttpServletResponse response, Request proxyRequest, InputStream input)        {            super(input);            this.request = request;            this.response = response;            this.proxyRequest = proxyRequest;        }        @Override        public long getLength()        {            return request.getContentLength();        }        @Override        protected ByteBuffer onRead(byte[] buffer, int offset, int length)        {            if (_log.isDebugEnabled())                _log.debug("{} proxying content to upstream: {} bytes", getRequestId(request), length);            return onRequestContent(request, proxyRequest, buffer, offset, length);        }        protected ByteBuffer onRequestContent(HttpServletRequest request, Request proxyRequest, byte[] buffer, int offset, int length)        {            return super.onRead(buffer, offset, length);        }        @Override        protected void onReadFailure(Throwable failure)        {            onClientRequestFailure(request, proxyRequest, response, failure);        }    }}