Project: Jetty

Revision: 3598

Author: gregw

Date: 03 Sep 2008 22:36:16

• /jetty/trunk/modules/extra/cometd/bayeux/src/main/java/org/mortbay/cometd/continuation/ContinuationClient.java

  ......	@@ -65,8 +65,6 @@
  65	65	/* ------------------------------------------------------------ */
  66	66	public void setContinuation(Continuation continuation)
  67	67	{
  68		Timeout.Task task=null;
  69
  70	68	if (continuation==null)
  71	69	{
  72	70	synchronized (this)
  ......	@@ -77,11 +75,8 @@
  77	75	_continuation.resume();
  78	76	}
  79	77	_continuation=null;
  80		task=_timeout;
  	78	_bayeux.startTimeout(_timeout,getTimeout());
  81	79	}
  82
  83		if (task!=null)
  84		_bayeux.startTimeout(task,getTimeout());
  85	80	}
  86	81	else
  87	82	{
  ......	@@ -93,11 +88,8 @@
  93	88	_continuation.resume();
  94	89	}
  95	90	_continuation=continuation;
  96		task=_timeout;
  	91	_bayeux.cancelTimeout(_timeout);
  97	92	}
  98
  99		if (task!=null)
  100		_bayeux.cancelTimeout(task);
  101	93	}
  102	94	}
  103	95
  ......	@@ -111,19 +103,14 @@
  111	103	@Override
  112	104	public void resume()
  113	105	{
  114		Timeout.Task task=null;
  115	106	synchronized (this)
  116	107	{
  117	108	if (_continuation!=null)
  118	109	{
  119	110	_continuation.resume();
  120		task=_timeout;
  121	111	}
  122	112	_continuation=null;
  123		}
  124
  125		if (task!=null)
  126		_bayeux.startTimeout(task,getTimeout());
  	113	}
  127	114	}
  128	115
  129	116	/* ------------------------------------------------------------ */
  ......	@@ -136,17 +123,15 @@
  136	123	/* ------------------------------------------------------------ */
  137	124	public void access()
  138	125	{
  139		Timeout.Task task=null;
  140	126	synchronized(this)
  141	127	{
  142	128	// distribute access time in cluster
  143	129	_accessed=_bayeux.getNow();
  144	130	if (_timeout!=null && _timeout.isScheduled())
  145		task=_timeout;
  	131	{
  	132	_timeout.reschedule();
  	133	}
  146	134	}
  147
  148		if (task!=null)
  149		_bayeux.startTimeout(task,getTimeout());
  150	135	}
  151	136
  152	137
  ......	@@ -163,18 +148,13 @@
  163	148	@Override
  164	149	public void remove(boolean wasTimeout)
  165	150	{
  166		Timeout.Task task=null;
  167	151	synchronized(this)
  168	152	{
  169		if (!wasTimeout)
  170		task=_timeout;
  	153	if (!wasTimeout && _timeout!=null)
  	154	_bayeux.cancelTimeout(_timeout);
  171	155	_timeout=null;
  172	156	super.remove(wasTimeout);
  173		}
  174
  175		if (task!=null)
  176		_bayeux.cancelTimeout(task);
  177
  	157	}
  178	158	}
  179	159
  180	160	}
  181	161	\ No newline at end of file

• /jetty/trunk/modules/extra/cometd/bayeux/src/main/java/org/mortbay/cometd/continuation/ContinuationCometdServlet.java

  ......	@@ -15,7 +15,6 @@
  15	15	package org.mortbay.cometd.continuation;
  16	16
  17	17	import java.io.IOException;
  18		import java.io.UnsupportedEncodingException;
  19	18	import java.nio.ByteBuffer;
  20	19
  21	20	import javax.servlet.ServletException;
  ......	@@ -27,12 +26,12 @@
  27	26	import org.cometd.Message;
  28	27	import org.mortbay.cometd.AbstractBayeux;
  29	28	import org.mortbay.cometd.AbstractCometdServlet;
  30		import org.mortbay.cometd.AbstractTransport;
  31	29	import org.mortbay.cometd.ClientImpl;
  32	30	import org.mortbay.cometd.JSONTransport;
  33	31	import org.mortbay.cometd.MessageImpl;
  34	32	import org.mortbay.cometd.Transport;
  35	33	import org.mortbay.util.ArrayQueue;
  	34	import org.mortbay.util.StringUtil;
  36	35	import org.mortbay.util.ajax.Continuation;
  37	36	import org.mortbay.util.ajax.ContinuationSupport;
  38	37
  ......	@@ -60,7 +59,7 @@
  60	59	ContinuationClient client=(clientObj instanceof ClientImpl)?(ContinuationClient)clientObj:null;
  61	60	Transport transport=null;
  62	61	boolean connect=false;
  63		int num_msgs=-1;
  	62	int received=-1;
  64	63
  65	64	// Have we seen this request before
  66	65	if (client!=null)
  ......	@@ -72,7 +71,7 @@
  72	71	else
  73	72	{
  74	73	Message[] messages = getMessages(request);
  75		num_msgs=messages.length;
  	74	received=messages.length;
  76	75
  77	76	/* check jsonp parameter */
  78	77	String jsonpParam=request.getParameter("jsonp");
  ......	@@ -140,10 +139,11 @@
  140	139	}
  141	140	}
  142	141
  	142	Message pollReply=null;
  143	143	// Do we need to wait for messages
  144	144	if (transport!=null)
  145	145	{
  146		Message pollReply=transport.getPollReply();
  	146	pollReply=transport.getPollReply();
  147	147	if (pollReply!=null)
  148	148	{
  149	149	if (_bayeux.isLogDebug())
  ......	@@ -153,13 +153,11 @@
  153	153	timeout=_bayeux.getTimeout();
  154	154
  155	155	Continuation continuation=ContinuationSupport.getContinuation(request,client);
  156		if (!continuation.isPending())
  157		client.access();
  158	156
  159	157	// Get messages or wait
  160	158	synchronized (client)
  161	159	{
  162		if (!client.hasMessages() && !continuation.isPending()&& num_msgs<=1)
  	160	if (!client.hasMessages() && !continuation.isPending()&& received<=1)
  163	161	{
  164	162	// save state and suspend
  165	163	((ContinuationClient)client).setContinuation(continuation);
  ......	@@ -167,6 +165,10 @@
  167	165	request.setAttribute(TRANSPORT_ATTR,transport);
  168	166	continuation.suspend(timeout);
  169	167	}
  	168
  	169	if (!continuation.isPending())
  	170	client.access();
  	171
  170	172	continuation.reset();
  171	173	}
  172	174
  ......	@@ -175,7 +177,6 @@
  175	177
  176	178	for (Extension e:_bayeux.getExtensions())
  177	179	pollReply=e.sendMeta(pollReply);
  178		transport.send(pollReply);
  179	180	}
  180	181	else if (client!=null)
  181	182	{
  ......	@@ -195,53 +196,57 @@
  195	196	boolean flushed=false;
  196	197	try
  197	198	{
  198		if (size==1)
  	199	if (pollReply!=null)
  199	200	{
  200		MessageImpl message = (MessageImpl)messages.peek();
  201
  202		ByteBuffer buffer = message.getBuffer();
  203		if (message.getRefs()>=_refsThreshold && transport instanceof JSONTransport)
  	201	// can we bypass response generation?
  	202	if (_refsThreshold>0 && size==1 && transport instanceof JSONTransport)
  204	203	{
  205		if (buffer==null)
  206		{
  207		// create a new buffer
  208		AbstractTransport trans = (AbstractTransport)transport;
  209		MessageImpl connectResponse = new MessageImpl();
  210		connectResponse.put(Bayeux.SUCCESSFUL_FIELD,Boolean.TRUE);
  211		connectResponse.put(Bayeux.CHANNEL_FIELD,Bayeux.META_CONNECT);
  212
  213		byte[] contentBytes = ("["+connectResponse.getJSON()+","+message.getJSON()+"]").getBytes();
  	204	MessageImpl message = (MessageImpl)messages.peek();
  	205
  	206	// is there a response already prepared
  	207	ByteBuffer buffer = message.getBuffer();
  	208	if (buffer!=null)
  	209	{
  	210	synchronized (buffer)
  	211	{
  	212	request.setAttribute("org.mortbay.jetty.ResponseBuffer",buffer);
  	213	((MessageImpl)message).decRef();
  	214	flushed=true;
  	215	}
  	216	}
  	217	else if (message.getRefs()>=_refsThreshold)
  	218	{
  	219	byte[] contentBytes = ("[{\""+Bayeux.SUCCESSFUL_FIELD+"\":true,\""+
  	220	Bayeux.CHANNEL_FIELD+"\":\""+Bayeux.META_CONNECT+"\"},"+
  	221	message.getJSON()+"]").getBytes(StringUtil.__UTF8);
  214	222	int contentLength = contentBytes.length;
  215	223
  216	224	String headerString = "HTTP/1.1 200 OK\r\n"+
  217		"Content-Type: text/json; charset=utf-8\r\n" +
  218		"Content-Length: " + contentLength + "\r\n" +
  219		"\r\n";
  	225	"Content-Type: text/json; charset=utf-8\r\n" +
  	226	"Content-Length: " + contentLength + "\r\n" +
  	227	"\r\n";
  220	228
  221		byte[] headerBytes = null;
  222		try
  223		{
  224		headerBytes = headerString.getBytes(request.getCharacterEncoding());
  225		}
  226		catch (UnsupportedEncodingException e)
  227		{
  228		headerBytes = headerString.getBytes();
  229		}
  	229	byte[] headerBytes = headerString.getBytes(StringUtil.__UTF8);
  230	230
  231	231	buffer = ByteBuffer.allocateDirect(headerBytes.length+contentLength);
  232	232	buffer.put(headerBytes);
  233	233	buffer.put(contentBytes);
  234
  235		message.setBuffer(buffer);
  	234	buffer.flip();
  	235
  	236	synchronized (buffer)
  	237	{
  	238	message.setBuffer(buffer);
  	239
  	240	request.setAttribute("org.mortbay.jetty.ResponseBuffer",buffer);
  	241	((MessageImpl)message).decRef();
  	242	flushed=true;
  	243	}
  236	244	}
  	245	else
  	246	transport.send(pollReply);
  237	247	}
  238
  239		if (buffer != null)
  240		{
  241		request.setAttribute("org.mortbay.jetty.ResponseBuffer",buffer);
  242		((MessageImpl)message).decRef();
  243		flushed=true;
  244		}
  	248	else
  	249	transport.send(pollReply);
  245	250	}
  246	251
  247	252	if (!flushed)
  ......	@@ -254,7 +259,6 @@
  254	259	}
  255	260
  256	261	transport.complete();
  257		response.flushBuffer();
  258	262	flushed=true;
  259	263	}
  260	264	}
  ......	@@ -269,7 +273,8 @@
  269	273	client.resume();
  270	274	}
  271	275	else
  	276	{
  272	277	transport.complete();
  273
  	278	}
  274	279	}
  275	280	}

• /jetty/trunk/modules/server/jetty/src/main/java/org/mortbay/jetty/Request.java

  ......	@@ -46,7 +46,6 @@
  46	46
  47	47	import org.mortbay.io.Buffer;
  48	48	import org.mortbay.io.BufferUtil;
  49		import org.mortbay.io.ByteArrayBuffer;
  50	49	import org.mortbay.io.EndPoint;
  51	50	import org.mortbay.io.Portable;
  52	51	import org.mortbay.io.nio.NIOBuffer;
  ......	@@ -1107,8 +1106,6 @@
  1107	1106	synchronized (byteBuffer)
  1108	1107	{
  1109	1108	NIOBuffer buffer = new NIOBuffer(byteBuffer,true);
  1110		buffer.setGetIndex(byteBuffer.position());
  1111		buffer.setPutIndex(byteBuffer.limit());
  1112	1109	((HttpConnection.Output)getServletResponse().getOutputStream()).sendResponse(buffer);
  1113	1110	}
  1114	1111	}
  ......	@@ -1691,16 +1688,6 @@
  1691	1688	_queryEncoding=queryEncoding;
  1692	1689	_queryString=null;
  1693	1690	}
  1694
  1695		/* ------------------------------------------------------------ */
  1696		/**
  1697		* @param byteBuffer Content to flush directly to the
  1698		* {@link org.mortbay.io.EndPoint}
  1699		*/
  1700		private void flushResxponseBuffer(ByteBuffer byteBuffer)
  1701		{
  1702
  1703		}
  1704	1691
  1705	1692	/* ------------------------------------------------------------ */
  1706	1693	public void setRoleMap(Map map)