//========================================================================
//Copyright 2007 Mort Bay Consulting Pty. Ltd.
//------------------------------------------------------------------------
//Licensed under the Apache License, Version 2.0 (the "License");
//you may not use this file except in compliance with the License.
//You may obtain a copy of the License at
//http://www.apache.org/licenses/LICENSE-2.0
//Unless required by applicable law or agreed to in writing, software
//distributed under the License is distributed on an "AS IS" BASIS,
//WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//See the License for the specific language governing permissions and
//limitations under the License.
//========================================================================
package org.mortbay.cometd.continuation;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.cometd.Bayeux;
import org.cometd.Client;
import org.cometd.Message;
import org.mortbay.cometd.AbstractBayeux;
import org.mortbay.cometd.AbstractCometdServlet;
import org.mortbay.cometd.ClientImpl;
import org.mortbay.cometd.JSONTransport;
import org.mortbay.cometd.MessageImpl;
import org.mortbay.cometd.Transport;
import org.mortbay.util.ArrayQueue;
import org.mortbay.util.StringUtil;
import org.mortbay.util.ajax.Continuation;
import org.mortbay.util.ajax.ContinuationSupport;
public class ContinuationCometdServlet extends AbstractCometdServlet
{
public final static int __DEFAULT_REFS_THRESHOLD=0;
protected int _refsThreshold=__DEFAULT_REFS_THRESHOLD;
String _responseBuffer;
@Override
public void init() throws ServletException
{
String refsThreshold=getInitParameter("refsThreshold");
if (refsThreshold != null)
_refsThreshold=Integer.parseInt(refsThreshold);
if (_refsThreshold>0)
{
String server = getServletContext().getServerInfo();
if (server.startsWith("jetty/6"))
_responseBuffer="org.mortbay.jetty.ResponseBuffer";
else if (server.startsWith("jetty/"))
_responseBuffer="org.eclipse.jetty.server.ResponseBuffer";
else
_refsThreshold=0;
}
super.init();
}
/* ------------------------------------------------------------ */
@Override
protected AbstractBayeux newBayeux()
{
return new ContinuationBayeux();
}
/* ------------------------------------------------------------ */
@Override
protected void service(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException
{
if (response.isCommitted())
{
// return here as this is a resumed request that has already had a
// response sent - this is probably a sendError();resume(); used instead
// of a complete().
return;
}
// Look for an existing client and protect from context restarts
Object clientObj=request.getAttribute(CLIENT_ATTR);
Transport transport=null;
int received=-1;
boolean metaConnectDeliveryOnly=false;
boolean pendingResponse=false;
boolean metaConnect=false;
final boolean initial;
// Have we seen this request before
ContinuationClient client=(clientObj instanceof ClientImpl)?(ContinuationClient)clientObj:null;
if (client != null)
{
initial=false;
// yes - extract saved properties
transport=(Transport)request.getAttribute(TRANSPORT_ATTR);
transport.setResponse(response);
metaConnectDeliveryOnly=client.isMetaConnectDeliveryOnly() || transport.isMetaConnectDeliveryOnly();
metaConnect=true;
}
else
{
initial=true;
Message[] messages=getMessages(request);
received=messages.length;
/* check jsonp parameter */
String jsonpParam=request.getParameter("jsonp");
// Handle all received messages
try
{
for (Message message : messages)
{
if (jsonpParam != null)
message.put("jsonp",jsonpParam);
if (client == null)
{
String clientId = (String)message.get(AbstractBayeux.CLIENT_FIELD);
client=(ContinuationClient)_bayeux.getClient(clientId);
// If no client, SHOULD be a handshake, so force a
// transport and handle
if (client == null)
{
// Setup a browser ID
String browser_id=findBrowserId(request);
if (browser_id == null)
browser_id=setBrowserId(request,response);
if (transport == null)
{
transport=_bayeux.newTransport(client,message);
transport.setResponse(response);
metaConnectDeliveryOnly=transport.isMetaConnectDeliveryOnly();
}
_bayeux.handle(null,transport,message);
message=null;
continue;
}
}
String browser_id=findBrowserId(request);
if (browser_id != null && (client.getBrowserId() == null || !client.getBrowserId().equals(browser_id)))
client.setBrowserId(browser_id);
// resolve transport
if (transport == null)
{
transport=_bayeux.newTransport(client,message);
transport.setResponse(response);
metaConnectDeliveryOnly=client.isMetaConnectDeliveryOnly() || transport.isMetaConnectDeliveryOnly();
}
// Tell client to hold messages as a response is likely to
// be sent.
if (!metaConnectDeliveryOnly && !pendingResponse)
{
pendingResponse=true;
client.responsePending();
}
if (Bayeux.META_CONNECT.equals(message.getChannel()))
metaConnect=true;
_bayeux.handle(client,transport,message);
}
}
finally
{
for (Message message : messages)
((MessageImpl)message).decRef();
if (pendingResponse)
{
client.responded();
}
}
}
Message metaConnectReply=null;
// Do we need to wait for messages
if (transport != null)
{
metaConnectReply=transport.getMetaConnectReply();
if (metaConnectReply != null)
{
long timeout=client.getTimeout();
if (timeout < 0)
timeout=_bayeux.getTimeout();
Continuation continuation=ContinuationSupport.getContinuation(request,client);
// Get messages or wait
synchronized(client)
{
if (timeout > 0 && !client.hasNonLazyMessages() && initial && received <= 1)
{
// save state and suspend
request.setAttribute(CLIENT_ATTR,client);
request.setAttribute(TRANSPORT_ATTR,transport);
client.setContinuation(continuation);
continuation.setObject(response);
continuation.suspend(timeout);
}
continuation.reset();
}
client.setContinuation(null);
transport.setMetaConnectReply(null);
}
else if (client != null)
{
client.access();
}
}
if (client != null)
{
if (metaConnectDeliveryOnly && !metaConnect)
{
// wake up any long poll
client.resume();
}
else
{
List<Message> messages;
synchronized (client)
{
client.doDeliverListeners();
ArrayQueue<Message> clientMessages = (ArrayQueue<Message>)client.getQueue();
// Copy the messages to avoid synchronization
messages = new ArrayList<Message>(clientMessages);
// Empty client's queue
clientMessages.clear();
}
final int size=messages.size();
for (int i=0; i < size; i++)
{
final Message message=messages.get(i);
final MessageImpl mesgImpl=(message instanceof MessageImpl)?(MessageImpl)message:null;
// Can we short cut the message?
if (i == 0 && size == 1 && mesgImpl != null && _refsThreshold > 0 && metaConnectReply != null && transport instanceof JSONTransport)
{
// is there a response already prepared
ByteBuffer buffer=mesgImpl.getBuffer();
if (buffer != null)
{
// Send pre-prepared buffer
request.setAttribute("org.mortbay.jetty.ResponseBuffer",buffer);
if (metaConnectReply instanceof MessageImpl)
((MessageImpl)metaConnectReply).decRef();
metaConnectReply=null;
transport=null;
mesgImpl.decRef();
continue;
}
else if (mesgImpl.getRefs() >= _refsThreshold)
{
// create multi-use buffer
byte[] contentBytes=("[" + mesgImpl.getJSON() + ",{\"" + Bayeux.SUCCESSFUL_FIELD + "\":true,\"" + Bayeux.CHANNEL_FIELD
+ "\":\"" + Bayeux.META_CONNECT + "\"}]").getBytes(StringUtil.__UTF8);
int contentLength=contentBytes.length;
String headerString="HTTP/1.1 200 OK\r\n" + "Content-Type: text/json; charset=utf-8\r\n" + "Content-Length: "
+ contentLength + "\r\n" + "\r\n";
byte[] headerBytes=headerString.getBytes(StringUtil.__UTF8);
buffer=ByteBuffer.allocateDirect(headerBytes.length + contentLength);
buffer.put(headerBytes);
buffer.put(contentBytes);
buffer.flip();
mesgImpl.setBuffer(buffer);
request.setAttribute("org.mortbay.jetty.ResponseBuffer",buffer);
metaConnectReply=null;
if (metaConnectReply instanceof MessageImpl)
((MessageImpl)metaConnectReply).decRef();
transport=null;
mesgImpl.decRef();
continue;
}
}
if (message != null)
transport.send(message);
if (mesgImpl != null)
mesgImpl.decRef();
}
if (metaConnectReply != null)
{
metaConnectReply=_bayeux.extendSendMeta(client,metaConnectReply);
transport.send(metaConnectReply);
if (metaConnectReply instanceof MessageImpl)
((MessageImpl)metaConnectReply).decRef();
}
}
}
if (transport != null)
transport.complete();
}
public void destroy()
{
ContinuationBayeux bayeux = (ContinuationBayeux)_bayeux;
if (bayeux != null)
{
for (Client c : bayeux.getClients())
{
if (c instanceof ContinuationClient)
{
ContinuationClient client = (ContinuationClient)c;
client.setContinuation(null);
}
}
bayeux.destroy();
}
}
}