();
-
- // Set the starting ciphers - either from the included or enabled list
- if (_includeCipherSuites!=null)
- {
- // Use only the supported included ciphers
- for (String cipherSuite : supportedCipherSuites)
- if (_includeCipherSuites.contains(cipherSuite))
- selected_ciphers.add(cipherSuite);
- }
- else
- selected_ciphers.addAll(Arrays.asList(enabledCipherSuites));
-
-
- // Remove any excluded ciphers
- if (_excludeCipherSuites != null)
- selected_ciphers.removeAll(_excludeCipherSuites);
- return selected_ciphers.toArray(new String[selected_ciphers.size()]);
- }
-
- /* ------------------------------------------------------------ */
- /**
- * Check if the lifecycle has been started and throw runtime exception
- */
- protected void checkNotStarted()
- {
- if (isStarted())
- throw new IllegalStateException("Cannot modify configuration when "+getState());
- }
-
- /* ------------------------------------------------------------ */
- /**
- * @return true if CRL Distribution Points support is enabled
- */
- public boolean isEnableCRLDP()
- {
- return _enableCRLDP;
- }
-
- /* ------------------------------------------------------------ */
- /** Enables CRL Distribution Points Support
- * @param enableCRLDP true - turn on, false - turns off
- */
- public void setEnableCRLDP(boolean enableCRLDP)
- {
- checkNotStarted();
-
- _enableCRLDP = enableCRLDP;
- }
-
- /* ------------------------------------------------------------ */
- /**
- * @return true if On-Line Certificate Status Protocol support is enabled
- */
- public boolean isEnableOCSP()
- {
- return _enableOCSP;
- }
-
- /* ------------------------------------------------------------ */
- /** Enables On-Line Certificate Status Protocol support
- * @param enableOCSP true - turn on, false - turn off
- */
- public void setEnableOCSP(boolean enableOCSP)
- {
- checkNotStarted();
-
- _enableOCSP = enableOCSP;
- }
-
- /* ------------------------------------------------------------ */
- /**
- * @return Location of the OCSP Responder
- */
- public String getOcspResponderURL()
- {
- return _ocspResponderURL;
- }
-
- /* ------------------------------------------------------------ */
- /** Set the location of the OCSP Responder.
- * @param ocspResponderURL location of the OCSP Responder
- */
- public void setOcspResponderURL(String ocspResponderURL)
- {
- checkNotStarted();
-
- _ocspResponderURL = ocspResponderURL;
- }
-
- /* ------------------------------------------------------------ */
- /** Set the key store.
- * @param keyStore the key store to set
- */
- public void setKeyStore(KeyStore keyStore)
- {
- checkNotStarted();
-
- _keyStore = keyStore;
- }
-
- /* ------------------------------------------------------------ */
- /** Set the trust store.
- * @param trustStore the trust store to set
- */
- public void setTrustStore(KeyStore trustStore)
- {
- checkNotStarted();
-
- _trustStore = trustStore;
- }
-
- /* ------------------------------------------------------------ */
- /** Set the key store resource.
- * @param resource the key store resource to set
- */
- public void setKeyStoreResource(Resource resource)
- {
- checkNotStarted();
-
- try
- {
- _keyStoreInputStream = resource.getInputStream();
- }
- catch (IOException e)
- {
- throw new InvalidParameterException("Unable to get resource "+
- "input stream for resource "+resource.toString());
- }
- }
-
- /* ------------------------------------------------------------ */
- /** Set the trust store resource.
- * @param resource the trust store resource to set
- */
- public void setTrustStore(Resource resource)
- {
- checkNotStarted();
-
- try
- {
- _trustStoreInputStream = resource.getInputStream();
- }
- catch (IOException e)
- {
- throw new InvalidParameterException("Unable to get resource "+
- "input stream for resource "+resource.toString());
- }
- }
-
- /* ------------------------------------------------------------ */
- /**
- * @return true if SSL Session caching is enabled
- */
- public boolean isSessionCachingEnabled()
- {
- return _sessionCachingEnabled;
- }
-
- /* ------------------------------------------------------------ */
- /** Set the flag to enable SSL Session caching.
- * @param enableSessionCaching the value of the flag
- */
- public void setSessionCachingEnabled(boolean enableSessionCaching)
- {
- _sessionCachingEnabled = enableSessionCaching;
- }
-
- /* ------------------------------------------------------------ */
- /** Get SSL session cache size.
- * @return SSL session cache size
- */
- public int getSslSessionCacheSize()
- {
- return _sslSessionCacheSize;
- }
-
- /* ------------------------------------------------------------ */
- /** SEt SSL session cache size.
- * @param sslSessionCacheSize SSL session cache size to set
- */
- public void setSslSessionCacheSize(int sslSessionCacheSize)
- {
- _sslSessionCacheSize = sslSessionCacheSize;
- }
-
- /* ------------------------------------------------------------ */
- /** Get SSL session timeout.
- * @return SSL session timeout
- */
- public int getSslSessionTimeout()
- {
- return _sslSessionTimeout;
- }
-
- /* ------------------------------------------------------------ */
- /** Set SSL session timeout.
- * @param sslSessionTimeout SSL session timeout to set
- */
- public void setSslSessionTimeout(int sslSessionTimeout)
- {
- _sslSessionTimeout = sslSessionTimeout;
- }
-
-
- /* ------------------------------------------------------------ */
- public SSLServerSocket newSslServerSocket(String host,int port,int backlog) throws IOException
- {
- SSLServerSocketFactory factory = _context.getServerSocketFactory();
-
- SSLServerSocket socket =
- (SSLServerSocket) (host==null ?
- factory.createServerSocket(port,backlog):
- factory.createServerSocket(port,backlog,InetAddress.getByName(host)));
-
- if (getWantClientAuth())
- socket.setWantClientAuth(getWantClientAuth());
- if (getNeedClientAuth())
- socket.setNeedClientAuth(getNeedClientAuth());
-
- socket.setEnabledCipherSuites(selectCipherSuites(
- socket.getEnabledCipherSuites(),
- socket.getSupportedCipherSuites()));
- socket.setEnabledProtocols(selectProtocols(socket.getEnabledProtocols(),socket.getSupportedProtocols()));
-
- return socket;
- }
-
- /* ------------------------------------------------------------ */
- public SSLSocket newSslSocket() throws IOException
- {
- SSLSocketFactory factory = _context.getSocketFactory();
-
- SSLSocket socket = (SSLSocket)factory.createSocket();
-
- if (getWantClientAuth())
- socket.setWantClientAuth(getWantClientAuth());
- if (getNeedClientAuth())
- socket.setNeedClientAuth(getNeedClientAuth());
-
- socket.setEnabledCipherSuites(selectCipherSuites(
- socket.getEnabledCipherSuites(),
- socket.getSupportedCipherSuites()));
- socket.setEnabledProtocols(selectProtocols(socket.getEnabledProtocols(),socket.getSupportedProtocols()));
-
- return socket;
- }
-
- /* ------------------------------------------------------------ */
- public SSLEngine newSslEngine(String host,int port)
- {
- SSLEngine sslEngine=isSessionCachingEnabled()
- ?_context.createSSLEngine(host, port)
- :_context.createSSLEngine();
-
- customize(sslEngine);
- return sslEngine;
- }
-
- /* ------------------------------------------------------------ */
- public SSLEngine newSslEngine()
- {
- SSLEngine sslEngine=_context.createSSLEngine();
- customize(sslEngine);
- return sslEngine;
- }
-
- /* ------------------------------------------------------------ */
- public void customize(SSLEngine sslEngine)
- {
- if (getWantClientAuth())
- sslEngine.setWantClientAuth(getWantClientAuth());
- if (getNeedClientAuth())
- sslEngine.setNeedClientAuth(getNeedClientAuth());
-
- sslEngine.setEnabledCipherSuites(selectCipherSuites(
- sslEngine.getEnabledCipherSuites(),
- sslEngine.getSupportedCipherSuites()));
-
- sslEngine.setEnabledProtocols(selectProtocols(sslEngine.getEnabledProtocols(),sslEngine.getSupportedProtocols()));
- }
-
}
diff --git a/jetty-http/src/test/java/org/eclipse/jetty/http/HttpFieldsTest.java b/jetty-http/src/test/java/org/eclipse/jetty/http/HttpFieldsTest.java
index 6120f942aec..3ae9a409fd0 100644
--- a/jetty-http/src/test/java/org/eclipse/jetty/http/HttpFieldsTest.java
+++ b/jetty-http/src/test/java/org/eclipse/jetty/http/HttpFieldsTest.java
@@ -14,10 +14,9 @@
package org.eclipse.jetty.http;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.assertFalse;
import java.util.Enumeration;
import java.util.HashSet;
diff --git a/jetty-http/src/test/java/org/eclipse/jetty/http/HttpGeneratorClientTest.java b/jetty-http/src/test/java/org/eclipse/jetty/http/HttpGeneratorClientTest.java
index 46841564daa..18012e950f5 100644
--- a/jetty-http/src/test/java/org/eclipse/jetty/http/HttpGeneratorClientTest.java
+++ b/jetty-http/src/test/java/org/eclipse/jetty/http/HttpGeneratorClientTest.java
@@ -138,7 +138,7 @@ public class HttpGeneratorClientTest
String t="v="+v+",r="+r+",chunks="+chunks+",c="+c+",tr="+tr[r];
// System.err.println(t);
- hb.reset(true);
+ hb.reset();
endp.reset();
fields.clear();
@@ -193,7 +193,7 @@ public class HttpGeneratorClientTest
}
private static final String[] headers= { "Content-Type","Content-Length","Connection","Transfer-Encoding","Other"};
- private class TR
+ private static class TR
{
private String[] values=new String[headers.length];
private String body;
diff --git a/jetty-http/src/test/java/org/eclipse/jetty/http/HttpGeneratorTest.java b/jetty-http/src/test/java/org/eclipse/jetty/http/HttpGeneratorTest.java
index 547d4647bb9..bceb1541e9d 100644
--- a/jetty-http/src/test/java/org/eclipse/jetty/http/HttpGeneratorTest.java
+++ b/jetty-http/src/test/java/org/eclipse/jetty/http/HttpGeneratorTest.java
@@ -86,7 +86,7 @@ public class HttpGeneratorTest
String t="v="+v+",r="+r+",chunks="+chunks+",connect="+c+",tr="+tr[r];
// System.err.println(t);
- hb.reset(true);
+ hb.reset();
endp.reset();
fields.clear();
@@ -138,7 +138,7 @@ public class HttpGeneratorTest
}
private static final String[] headers= { "Content-Type","Content-Length","Connection","Transfer-Encoding","Other"};
- private class TR
+ private static class TR
{
private int _code;
private String _body;
diff --git a/jetty-http/src/test/java/org/eclipse/jetty/http/HttpParserTest.java b/jetty-http/src/test/java/org/eclipse/jetty/http/HttpParserTest.java
index ac4ae8a1552..85cf15c202c 100644
--- a/jetty-http/src/test/java/org/eclipse/jetty/http/HttpParserTest.java
+++ b/jetty-http/src/test/java/org/eclipse/jetty/http/HttpParserTest.java
@@ -210,6 +210,7 @@ public class HttpParserTest
StringEndPoint io=new StringEndPoint();
io.setInput(
"GET /mp HTTP/1.0\015\012"
+ + "Connection: Keep-Alive\015\012"
+ "Header1: value1\015\012"
+ "Transfer-Encoding: chunked\015\012"
+ "\015\012"
@@ -219,10 +220,12 @@ public class HttpParserTest
+ "ABCDEFGHIJKLMNOPQRSTUVWXYZ\015\012"
+ "0\015\012"
+ "POST /foo HTTP/1.0\015\012"
+ + "Connection: Keep-Alive\015\012"
+ "Header2: value2\015\012"
+ "Content-Length: 0\015\012"
+ "\015\012"
+ "PUT /doodle HTTP/1.0\015\012"
+ + "Connection: close\015\012"
+ "Header3: value3\015\012"
+ "Content-Length: 10\015\012"
+ "\015\012"
@@ -238,27 +241,27 @@ public class HttpParserTest
assertEquals("GET", f0);
assertEquals("/mp", f1);
assertEquals("HTTP/1.0", f2);
- assertEquals(1, h);
- assertEquals("Header1", hdr[0]);
- assertEquals("value1", val[0]);
+ assertEquals(2, h);
+ assertEquals("Header1", hdr[1]);
+ assertEquals("value1", val[1]);
assertEquals("0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ", _content);
parser.parse();
assertEquals("POST", f0);
assertEquals("/foo", f1);
assertEquals("HTTP/1.0", f2);
- assertEquals(1, h);
- assertEquals("Header2", hdr[0]);
- assertEquals("value2", val[0]);
+ assertEquals(2, h);
+ assertEquals("Header2", hdr[1]);
+ assertEquals("value2", val[1]);
assertEquals(null, _content);
parser.parse();
assertEquals("PUT", f0);
assertEquals("/doodle", f1);
assertEquals("HTTP/1.0", f2);
- assertEquals(1, h);
- assertEquals("Header3", hdr[0]);
- assertEquals("value3", val[0]);
+ assertEquals(2, h);
+ assertEquals("Header3", hdr[1]);
+ assertEquals("value3", val[1]);
assertEquals("0123456789", _content);
}
@@ -266,7 +269,8 @@ public class HttpParserTest
public void testStreamParse() throws Exception
{
StringEndPoint io=new StringEndPoint();
- String http="GET / HTTP/1.0\015\012"
+ String http="GET / HTTP/1.1\015\012"
+ + "Host: test\015\012"
+ "Header1: value1\015\012"
+ "Transfer-Encoding: chunked\015\012"
+ "\015\012"
@@ -275,11 +279,14 @@ public class HttpParserTest
+ "1a\015\012"
+ "ABCDEFGHIJKLMNOPQRSTUVWXYZ\015\012"
+ "0\015\012"
- + "POST /foo HTTP/1.0\015\012"
+ + "POST /foo HTTP/1.1\015\012"
+ + "Host: test\015\012"
+ "Header2: value2\015\012"
+ "Content-Length: 0\015\012"
+ "\015\012"
- + "PUT /doodle HTTP/1.0\015\012"
+ + "PUT /doodle HTTP/1.1\015\012"
+ + "Host: test\015\012"
+ + "Connection: close\015\012"
+ "Header3: value3\015\012"
+ "Content-Length: 10\015\012"
+ "\015\012"
@@ -296,15 +303,17 @@ public class HttpParserTest
http.length() - 2,
http.length() / 2,
http.length() / 3,
- 64,
+ 128,
32
};
for (int t= 0; t < tests.length; t++)
{
- String tst="t"+tests[t];
+ String tst="t"+t+"="+tests[t];
try
- {
+ {
+ f0=f1=f2=null;
+ h=0;
ByteArrayBuffer buffer= new ByteArrayBuffer(tests[t]);
ByteArrayBuffer content=new ByteArrayBuffer(8192);
SimpleBuffers buffers=new SimpleBuffers(buffer,content);
@@ -314,31 +323,32 @@ public class HttpParserTest
io.setInput(http);
+ // System.err.println(tst);
parser.parse();
assertEquals(tst,"GET", f0);
assertEquals(tst,"/", f1);
- assertEquals(tst,"HTTP/1.0", f2);
- assertEquals(tst,1, h);
- assertEquals(tst,"Header1", hdr[0]);
- assertEquals(tst,"value1", val[0]);
+ assertEquals(tst,"HTTP/1.1", f2);
+ assertEquals(tst,2, h);
+ assertEquals(tst,"Header1", hdr[1]);
+ assertEquals(tst,"value1", val[1]);
assertEquals(tst,"0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ", _content);
parser.parse();
assertEquals(tst,"POST", f0);
assertEquals(tst,"/foo", f1);
- assertEquals(tst,"HTTP/1.0", f2);
- assertEquals(tst,1, h);
- assertEquals(tst,"Header2", hdr[0]);
- assertEquals(tst,"value2", val[0]);
+ assertEquals(tst,"HTTP/1.1", f2);
+ assertEquals(tst,2, h);
+ assertEquals(tst,"Header2", hdr[1]);
+ assertEquals(tst,"value2", val[1]);
assertEquals(tst,null, _content);
parser.parse();
assertEquals(tst,"PUT", f0);
assertEquals(tst,"/doodle", f1);
- assertEquals(tst,"HTTP/1.0", f2);
- assertEquals(tst,1, h);
- assertEquals(tst,"Header3", hdr[0]);
- assertEquals(tst,"value3", val[0]);
+ assertEquals(tst,"HTTP/1.1", f2);
+ assertEquals(tst,3, h);
+ assertEquals(tst,"Header3", hdr[2]);
+ assertEquals(tst,"value3", val[2]);
assertEquals(tst,"0123456789", _content);
}
catch(Exception e)
@@ -401,7 +411,7 @@ public class HttpParserTest
StringEndPoint io=new StringEndPoint();
io.setInput(
"HTTP/1.1 204 No-Content\015\012"
- + "Connection: close\015\012"
+ + "Header: value\015\012"
+ "\015\012"
+ "HTTP/1.1 200 Correct\015\012"
+ "Content-Length: 10\015\012"
diff --git a/jetty-http/src/test/java/org/eclipse/jetty/http/SslContextFactoryTest.java b/jetty-http/src/test/java/org/eclipse/jetty/http/SslContextFactoryTest.java
deleted file mode 100644
index f7fce832c0e..00000000000
--- a/jetty-http/src/test/java/org/eclipse/jetty/http/SslContextFactoryTest.java
+++ /dev/null
@@ -1,75 +0,0 @@
-package org.eclipse.jetty.http;
-
-import static junit.framework.Assert.assertTrue;
-
-import java.io.FileInputStream;
-import java.security.KeyStore;
-
-import org.eclipse.jetty.http.ssl.SslContextFactory;
-import org.junit.Test;
-
-
-public class SslContextFactoryTest
-{
- @Test
- public void testNoTsFileKs() throws Exception
- {
- String keystorePath = System.getProperty("basedir",".") + "/src/test/resources/keystore";
- SslContextFactory cf = new SslContextFactory(keystorePath);
- cf.setKeyStorePassword("storepwd");
- cf.setKeyManagerPassword("keypwd");
-
- cf.start();
-
- assertTrue(cf.getSslContext()!=null);
- }
-
- @Test
- public void testNoTsStreamKs() throws Exception
- {
- String keystorePath = System.getProperty("basedir",".") + "/src/test/resources/keystore";
-
- SslContextFactory cf = new SslContextFactory();
-
- cf.setKeyStoreInputStream(new FileInputStream(keystorePath));
- cf.setKeyStorePassword("storepwd");
- cf.setKeyManagerPassword("keypwd");
-
- cf.start();
-
- assertTrue(cf.getSslContext()!=null);
- }
-
- @Test
- public void testNoTsSetKs() throws Exception
- {
- String keystorePath = System.getProperty("basedir",".") + "/src/test/resources/keystore";
-
- KeyStore ks = KeyStore.getInstance("JKS");
- ks.load(new FileInputStream(keystorePath),"storepwd".toCharArray());
-
- SslContextFactory cf = new SslContextFactory();
- cf.setKeyStore(ks);
- cf.setKeyManagerPassword("keypwd");
-
- cf.start();
-
- assertTrue(cf.getSslContext()!=null);
- }
-
- @Test
- public void testNoTsNoKs() throws Exception
- {
- SslContextFactory cf = new SslContextFactory();
- cf.start();
- assertTrue(cf.getSslContext()!=null);
- }
-
- @Test
- public void testTrustAll() throws Exception
- {
- SslContextFactory cf = new SslContextFactory();
- cf.start();
- assertTrue(cf.getSslContext()!=null);
- }
-}
diff --git a/jetty-io/pom.xml b/jetty-io/pom.xml
index f4c816d3f27..8a5f69852a6 100644
--- a/jetty-io/pom.xml
+++ b/jetty-io/pom.xml
@@ -2,7 +2,7 @@
jetty-project
org.eclipse.jetty
- 7.5.4-SNAPSHOT
+ 7.6.0-SNAPSHOT
4.0.0
org.eclipse.jetty
diff --git a/jetty-io/src/main/java/org/eclipse/jetty/io/AbstractConnection.java b/jetty-io/src/main/java/org/eclipse/jetty/io/AbstractConnection.java
index 04a6e26ec87..a9167eba32e 100644
--- a/jetty-io/src/main/java/org/eclipse/jetty/io/AbstractConnection.java
+++ b/jetty-io/src/main/java/org/eclipse/jetty/io/AbstractConnection.java
@@ -15,13 +15,13 @@ public abstract class AbstractConnection implements Connection
public AbstractConnection(EndPoint endp)
{
- _endp=endp;
+ _endp=(EndPoint)endp;
_timeStamp = System.currentTimeMillis();
}
-
+
public AbstractConnection(EndPoint endp,long timestamp)
{
- _endp=endp;
+ _endp=(EndPoint)endp;
_timeStamp = timestamp;
}
@@ -29,17 +29,21 @@ public abstract class AbstractConnection implements Connection
{
return _timeStamp;
}
-
+
public EndPoint getEndPoint()
{
return _endp;
}
- public void idleExpired()
+ public void onIdleExpired(long idleForMs)
{
try
{
- _endp.shutdownOutput();
+ LOG.debug("onIdleExpired {}ms {} {}",idleForMs,this,_endp);
+ if (_endp.isInputShutdown() || _endp.isOutputShutdown())
+ _endp.close();
+ else
+ _endp.shutdownOutput();
}
catch(IOException e)
{
@@ -52,13 +56,18 @@ public abstract class AbstractConnection implements Connection
catch(IOException e2)
{
LOG.ignore(e2);
-
}
}
}
-
+
public String toString()
{
- return super.toString()+"@"+_endp.getLocalAddr()+":"+_endp.getLocalPort()+"<->"+_endp.getRemoteAddr()+":"+_endp.getRemotePort();
+ return String.format("%s@%x//%s:%d<->%s:%d",
+ getClass().getSimpleName(),
+ hashCode(),
+ _endp.getLocalAddr(),
+ _endp.getLocalPort(),
+ _endp.getRemoteAddr(),
+ _endp.getRemotePort());
}
}
diff --git a/jetty-io/src/main/java/org/eclipse/jetty/io/AsyncEndPoint.java b/jetty-io/src/main/java/org/eclipse/jetty/io/AsyncEndPoint.java
index 5eebc36ad80..992ad7a195a 100644
--- a/jetty-io/src/main/java/org/eclipse/jetty/io/AsyncEndPoint.java
+++ b/jetty-io/src/main/java/org/eclipse/jetty/io/AsyncEndPoint.java
@@ -13,36 +13,59 @@
package org.eclipse.jetty.io;
-public interface AsyncEndPoint extends EndPoint
+import org.eclipse.jetty.util.thread.Timeout;
+
+public interface AsyncEndPoint extends ConnectedEndPoint
{
/* ------------------------------------------------------------ */
/**
* Dispatch the endpoint to a thread to attend to it.
*
*/
- public void dispatch();
-
- /**
- * @return true if this endpoint can accept a dispatch. False if the
- * endpoint cannot accept a dispatched (eg is suspended or already dispatched)
- */
- public boolean isReadyForDispatch();
+ public void asyncDispatch();
/* ------------------------------------------------------------ */
/** Schedule a write dispatch.
* Set the endpoint to not be writable and schedule a dispatch when
* it becomes writable.
*/
- public void scheduleWrite();
-
- /* ------------------------------------------------------------ */
- /** Schedule a call to the idle timeout
- */
- public void scheduleIdle();
-
- /* ------------------------------------------------------------ */
- /** Cancel a call to the idle timeout
- */
- public void cancelIdle();
+ public void scheduleWrite();
+ /* ------------------------------------------------------------ */
+ /** Callback when idle.
+ * An endpoint is idle if there has been no IO activity for
+ * {@link #getMaxIdleTime()} and {@link #isCheckForIdle()} is true.
+ * @param idleForMs TODO
+ */
+ public void onIdleExpired(long idleForMs);
+
+ /* ------------------------------------------------------------ */
+ /** Set if the endpoint should be checked for idleness
+ */
+ public void setCheckForIdle(boolean check);
+
+ /* ------------------------------------------------------------ */
+ /** Get if the endpoint should be checked for idleness
+ */
+ public boolean isCheckForIdle();
+
+
+ /* ------------------------------------------------------------ */
+ public boolean isWritable();
+
+ /* ------------------------------------------------------------ */
+ /**
+ * @return True if IO has been successfully performed since the last call to {@link #hasProgressed()}
+ */
+ public boolean hasProgressed();
+
+ /* ------------------------------------------------------------ */
+ /**
+ */
+ public void scheduleTimeout(Timeout.Task task, long timeoutMs);
+
+ /* ------------------------------------------------------------ */
+ /**
+ */
+ public void cancelTimeout(Timeout.Task task);
}
diff --git a/jetty-io/src/main/java/org/eclipse/jetty/io/BufferCache.java b/jetty-io/src/main/java/org/eclipse/jetty/io/BufferCache.java
index 904961087dc..0abc1270182 100644
--- a/jetty-io/src/main/java/org/eclipse/jetty/io/BufferCache.java
+++ b/jetty-io/src/main/java/org/eclipse/jetty/io/BufferCache.java
@@ -103,6 +103,12 @@ public class BufferCache
return lookup(buffer).toString();
}
+ public int getOrdinal(String value)
+ {
+ CachedBuffer buffer = (CachedBuffer)_stringMap.get(value);
+ return buffer==null?-1:buffer.getOrdinal();
+ }
+
public int getOrdinal(Buffer buffer)
{
if (buffer instanceof CachedBuffer)
diff --git a/jetty-io/src/main/java/org/eclipse/jetty/io/BufferDateCache.java b/jetty-io/src/main/java/org/eclipse/jetty/io/BufferDateCache.java
index c1aebb88829..275d2f3b77d 100644
--- a/jetty-io/src/main/java/org/eclipse/jetty/io/BufferDateCache.java
+++ b/jetty-io/src/main/java/org/eclipse/jetty/io/BufferDateCache.java
@@ -46,7 +46,6 @@ public class BufferDateCache extends DateCache
public synchronized Buffer formatBuffer(long date)
{
String d = super.format(date);
- //noinspection StringEquality
if (d==_last)
return _buffer;
_last=d;
diff --git a/jetty-io/src/main/java/org/eclipse/jetty/io/ByteArrayEndPoint.java b/jetty-io/src/main/java/org/eclipse/jetty/io/ByteArrayEndPoint.java
index fcae2f99fda..a0e4f8e96df 100644
--- a/jetty-io/src/main/java/org/eclipse/jetty/io/ByteArrayEndPoint.java
+++ b/jetty-io/src/main/java/org/eclipse/jetty/io/ByteArrayEndPoint.java
@@ -4,11 +4,11 @@
// 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
+// 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.
+// You may elect to redistribute this code under either of these licenses.
// ========================================================================
package org.eclipse.jetty.io;
@@ -19,7 +19,7 @@ import java.io.IOException;
/* ------------------------------------------------------------ */
/** ByteArrayEndPoint.
- *
+ *
*
*/
public class ByteArrayEndPoint implements ConnectedEndPoint
@@ -36,12 +36,12 @@ public class ByteArrayEndPoint implements ConnectedEndPoint
/* ------------------------------------------------------------ */
/**
- *
+ *
*/
public ByteArrayEndPoint()
{
}
-
+
/* ------------------------------------------------------------ */
/**
* @see org.eclipse.jetty.io.ConnectedEndPoint#getConnection()
@@ -80,7 +80,7 @@ public class ByteArrayEndPoint implements ConnectedEndPoint
/* ------------------------------------------------------------ */
/**
- *
+ *
*/
public ByteArrayEndPoint(byte[] input, int outputSize)
{
@@ -121,9 +121,9 @@ public class ByteArrayEndPoint implements ConnectedEndPoint
{
_out = out;
}
-
+
/* ------------------------------------------------------------ */
- /*
+ /*
* @see org.eclipse.io.EndPoint#isOpen()
*/
public boolean isOpen()
@@ -150,7 +150,7 @@ public class ByteArrayEndPoint implements ConnectedEndPoint
}
/* ------------------------------------------------------------ */
- /*
+ /*
* @see org.eclipse.io.EndPoint#isBlocking()
*/
public boolean isBlocking()
@@ -171,7 +171,7 @@ public class ByteArrayEndPoint implements ConnectedEndPoint
}
/* ------------------------------------------------------------ */
- /*
+ /*
* @see org.eclipse.io.EndPoint#shutdownOutput()
*/
public void shutdownOutput() throws IOException
@@ -180,16 +180,16 @@ public class ByteArrayEndPoint implements ConnectedEndPoint
}
/* ------------------------------------------------------------ */
- /*
+ /*
* @see org.eclipse.io.EndPoint#shutdownInput()
*/
public void shutdownInput() throws IOException
{
close();
}
-
+
/* ------------------------------------------------------------ */
- /*
+ /*
* @see org.eclipse.io.EndPoint#close()
*/
public void close() throws IOException
@@ -198,30 +198,30 @@ public class ByteArrayEndPoint implements ConnectedEndPoint
}
/* ------------------------------------------------------------ */
- /*
+ /*
* @see org.eclipse.io.EndPoint#fill(org.eclipse.io.Buffer)
*/
public int fill(Buffer buffer) throws IOException
{
if (_closed)
throw new IOException("CLOSED");
-
+
if (_in!=null && _in.length()>0)
{
int len = buffer.put(_in);
_in.skip(len);
return len;
}
-
+
if (_in!=null && _in.length()==0 && _nonBlocking)
return 0;
-
+
close();
return -1;
}
/* ------------------------------------------------------------ */
- /*
+ /*
* @see org.eclipse.io.EndPoint#flush(org.eclipse.io.Buffer)
*/
public int flush(Buffer buffer) throws IOException
@@ -252,24 +252,24 @@ public class ByteArrayEndPoint implements ConnectedEndPoint
}
/* ------------------------------------------------------------ */
- /*
+ /*
* @see org.eclipse.io.EndPoint#flush(org.eclipse.io.Buffer, org.eclipse.io.Buffer, org.eclipse.io.Buffer)
*/
public int flush(Buffer header, Buffer buffer, Buffer trailer) throws IOException
{
if (_closed)
throw new IOException("CLOSED");
-
+
int flushed=0;
-
+
if (header!=null && header.length()>0)
flushed=flush(header);
-
+
if (header==null || header.length()==0)
{
if (buffer!=null && buffer.length()>0)
flushed+=flush(buffer);
-
+
if (buffer==null || buffer.length()==0)
{
if (trailer!=null && trailer.length()>0)
@@ -278,13 +278,13 @@ public class ByteArrayEndPoint implements ConnectedEndPoint
}
}
}
-
+
return flushed;
}
/* ------------------------------------------------------------ */
/**
- *
+ *
*/
public void reset()
{
@@ -296,7 +296,7 @@ public class ByteArrayEndPoint implements ConnectedEndPoint
}
/* ------------------------------------------------------------ */
- /*
+ /*
* @see org.eclipse.io.EndPoint#getLocalAddr()
*/
public String getLocalAddr()
@@ -305,7 +305,7 @@ public class ByteArrayEndPoint implements ConnectedEndPoint
}
/* ------------------------------------------------------------ */
- /*
+ /*
* @see org.eclipse.io.EndPoint#getLocalHost()
*/
public String getLocalHost()
@@ -314,7 +314,7 @@ public class ByteArrayEndPoint implements ConnectedEndPoint
}
/* ------------------------------------------------------------ */
- /*
+ /*
* @see org.eclipse.io.EndPoint#getLocalPort()
*/
public int getLocalPort()
@@ -323,7 +323,7 @@ public class ByteArrayEndPoint implements ConnectedEndPoint
}
/* ------------------------------------------------------------ */
- /*
+ /*
* @see org.eclipse.io.EndPoint#getRemoteAddr()
*/
public String getRemoteAddr()
@@ -332,7 +332,7 @@ public class ByteArrayEndPoint implements ConnectedEndPoint
}
/* ------------------------------------------------------------ */
- /*
+ /*
* @see org.eclipse.io.EndPoint#getRemoteHost()
*/
public String getRemoteHost()
@@ -341,7 +341,7 @@ public class ByteArrayEndPoint implements ConnectedEndPoint
}
/* ------------------------------------------------------------ */
- /*
+ /*
* @see org.eclipse.io.EndPoint#getRemotePort()
*/
public int getRemotePort()
@@ -350,7 +350,7 @@ public class ByteArrayEndPoint implements ConnectedEndPoint
}
/* ------------------------------------------------------------ */
- /*
+ /*
* @see org.eclipse.io.EndPoint#getConnection()
*/
public Object getTransport()
@@ -360,27 +360,9 @@ public class ByteArrayEndPoint implements ConnectedEndPoint
/* ------------------------------------------------------------ */
public void flush() throws IOException
- {
- }
-
- /* ------------------------------------------------------------ */
- public boolean isBufferingInput()
{
- return false;
}
-
- /* ------------------------------------------------------------ */
- public boolean isBufferingOutput()
- {
- return false;
- }
-
- /* ------------------------------------------------------------ */
- public boolean isBufferred()
- {
- return false;
- }
-
+
/* ------------------------------------------------------------ */
/**
* @return the growOutput
diff --git a/jetty-io/src/main/java/org/eclipse/jetty/io/Connection.java b/jetty-io/src/main/java/org/eclipse/jetty/io/Connection.java
index 431964afe8f..19cf3bee60e 100644
--- a/jetty-io/src/main/java/org/eclipse/jetty/io/Connection.java
+++ b/jetty-io/src/main/java/org/eclipse/jetty/io/Connection.java
@@ -50,10 +50,11 @@ public interface Connection
/**
* Called when the connection is closed
*/
- void closed();
+ void onClose();
/**
* Called when the connection idle timeout expires
+ * @param idleForMs TODO
*/
- void idleExpired();
+ void onIdleExpired(long idleForMs);
}
diff --git a/jetty-io/src/main/java/org/eclipse/jetty/io/EndPoint.java b/jetty-io/src/main/java/org/eclipse/jetty/io/EndPoint.java
index 6940991f387..231b752b9ee 100644
--- a/jetty-io/src/main/java/org/eclipse/jetty/io/EndPoint.java
+++ b/jetty-io/src/main/java/org/eclipse/jetty/io/EndPoint.java
@@ -4,11 +4,11 @@
// 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
+// 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.
+// You may elect to redistribute this code under either of these licenses.
// ========================================================================
package org.eclipse.jetty.io;
@@ -17,7 +17,7 @@ import java.io.IOException;
/**
- *
+ *
* A transport EndPoint
*/
public interface EndPoint
@@ -28,36 +28,36 @@ public interface EndPoint
void shutdownOutput() throws IOException;
boolean isOutputShutdown();
-
+
/**
* Shutdown any backing input stream associated with the endpoint
*/
void shutdownInput() throws IOException;
-
+
boolean isInputShutdown();
-
+
/**
* Close any backing stream associated with the endpoint
*/
void close() throws IOException;
/**
- * Fill the buffer from the current putIndex to it's capacity from whatever
+ * Fill the buffer from the current putIndex to it's capacity from whatever
* byte source is backing the buffer. The putIndex is increased if bytes filled.
* The buffer may chose to do a compact before filling.
- * @return an int value indicating the number of bytes
+ * @return an int value indicating the number of bytes
* filled or -1 if EOF is reached.
* @throws EofException If input is shutdown or the endpoint is closed.
*/
int fill(Buffer buffer) throws IOException;
-
+
/**
* Flush the buffer from the current getIndex to it's putIndex using whatever byte
* sink is backing the buffer. The getIndex is updated with the number of bytes flushed.
* Any mark set is cleared.
* If the entire contents of the buffer are flushed, then an implicit empty() is done.
- *
+ *
* @param buffer The buffer to flush. This buffers getIndex is updated.
* @return the number of bytes written
* @throws EofException If the endpoint is closed or output is shutdown.
@@ -69,7 +69,7 @@ public interface EndPoint
* sink is backing the buffer. The getIndex is updated with the number of bytes flushed.
* Any mark set is cleared.
* If the entire contents of the buffer are flushed, then an implicit empty() is done.
- * The passed header/trailer buffers are written before/after the contents of this buffer. This may be done
+ * The passed header/trailer buffers are written before/after the contents of this buffer. This may be done
* either as gather writes, as a poke into this buffer or as several writes. The implementation is free to
* select the optimal mechanism.
* @param header A buffer to write before flushing this buffer. This buffers getIndex is updated.
@@ -79,14 +79,14 @@ public interface EndPoint
*/
int flush(Buffer header, Buffer buffer, Buffer trailer) throws IOException;
-
+
/* ------------------------------------------------------------ */
/**
* @return The local IP address to which this EndPoint is bound, or null
* if this EndPoint does not represent a network connection.
*/
public String getLocalAddr();
-
+
/* ------------------------------------------------------------ */
/**
* @return The local host name to which this EndPoint is bound, or null
@@ -122,13 +122,9 @@ public interface EndPoint
*/
public int getRemotePort();
-
/* ------------------------------------------------------------ */
public boolean isBlocking();
-
- /* ------------------------------------------------------------ */
- public boolean isBufferred();
-
+
/* ------------------------------------------------------------ */
public boolean blockReadable(long millisecs) throws IOException;
@@ -143,27 +139,14 @@ public interface EndPoint
* @return The underlying transport object (socket, channel, etc.)
*/
public Object getTransport();
-
- /* ------------------------------------------------------------ */
- /**
- * @return True if the endpoint has some buffered input data
- */
- public boolean isBufferingInput();
-
- /* ------------------------------------------------------------ */
- /**
- * @return True if the endpoint has some buffered output data
- */
- public boolean isBufferingOutput();
-
+
/* ------------------------------------------------------------ */
/** Flush any buffered output.
* May fail to write all data if endpoint is non-blocking
* @throws EofException If the endpoint is closed or output is shutdown.
*/
public void flush() throws IOException;
-
-
+
/* ------------------------------------------------------------ */
/** Get the max idle time in ms.
*
The max idle time is the time the endpoint can be idle before
@@ -174,14 +157,14 @@ public interface EndPoint
* @return the max idle time in ms or if ms <= 0 implies an infinite timeout
*/
public int getMaxIdleTime();
-
+
/* ------------------------------------------------------------ */
/** Set the max idle time.
* @param timeMs the max idle time in MS. Timeout <= 0 implies an infinite timeout
* @throws IOException if the timeout cannot be set.
*/
public void setMaxIdleTime(int timeMs) throws IOException;
-
-
-
+
+
+
}
diff --git a/jetty-io/src/main/java/org/eclipse/jetty/io/EofException.java b/jetty-io/src/main/java/org/eclipse/jetty/io/EofException.java
index 9c864aa078a..03369e5a79d 100644
--- a/jetty-io/src/main/java/org/eclipse/jetty/io/EofException.java
+++ b/jetty-io/src/main/java/org/eclipse/jetty/io/EofException.java
@@ -15,6 +15,13 @@ package org.eclipse.jetty.io;
import java.io.EOFException;
+
+/* ------------------------------------------------------------ */
+/** A Jetty specialization of EOFException.
+ *
This is thrown by Jetty to distinguish between EOF received from
+ * the connection, vs and EOF thrown by some application talking to some other file/socket etc.
+ * The only difference in handling is that Jetty EOFs are logged less verbosely.
+ */
public class EofException extends EOFException
{
public EofException()
diff --git a/jetty-io/src/main/java/org/eclipse/jetty/io/UncheckedPrintWriter.java b/jetty-io/src/main/java/org/eclipse/jetty/io/UncheckedPrintWriter.java
index f114417e800..cfb05447a29 100644
--- a/jetty-io/src/main/java/org/eclipse/jetty/io/UncheckedPrintWriter.java
+++ b/jetty-io/src/main/java/org/eclipse/jetty/io/UncheckedPrintWriter.java
@@ -35,7 +35,8 @@ public class UncheckedPrintWriter extends PrintWriter
private static final Logger LOG = Log.getLogger(UncheckedPrintWriter.class);
private boolean _autoFlush = false;
- private boolean _throwUnchecked=true;
+ private IOException _ioException;
+ private boolean _isClosed = false;
/* ------------------------------------------------------------ */
/**
@@ -102,38 +103,45 @@ public class UncheckedPrintWriter extends PrintWriter
this(new BufferedWriter(new OutputStreamWriter(out)),autoFlush);
}
+
/* ------------------------------------------------------------ */
- private void setError(Throwable th)
+ public boolean checkError()
{
- setError();
- if (_throwUnchecked)
- throw new RuntimeIOException(th);
- LOG.debug(th);
- }
-
- /* ------------------------------------------------------------ */
- /** Are unchecked exceptions thrown.
- * @return True if {@link RuntimeIOException}s are thrown
- */
- public boolean isUncheckedPrintWriter()
- {
- return _throwUnchecked;
- }
-
- /* ------------------------------------------------------------ */
- /** Set if unchecked exceptions are thrown
- * @param uncheckedPrintWriter True if {@link RuntimeIOException}s are to be thrown
- */
- public void setUncheckedPrintWriter(boolean uncheckedPrintWriter)
- {
- _throwUnchecked = uncheckedPrintWriter;
+ return _ioException!=null || super.checkError();
}
+ /* ------------------------------------------------------------ */
+ private void setError(Throwable th)
+ {
+
+ super.setError();
+
+ if (th instanceof IOException)
+ _ioException=(IOException)th;
+ else
+ {
+ _ioException=new IOException(String.valueOf(th));
+ _ioException.initCause(th);
+ }
+
+ LOG.debug(th);
+ }
+
+
+ @Override
+ protected void setError()
+ {
+ setError(new IOException());
+ }
+
/* ------------------------------------------------------------ */
/** Check to make sure that the stream has not been closed */
private void isOpen() throws IOException
- {
- if (super.out == null)
+ {
+ if (_ioException!=null)
+ throw new RuntimeIOException(_ioException);
+
+ if (_isClosed)
throw new IOException("Stream closed");
}
@@ -170,6 +178,7 @@ public class UncheckedPrintWriter extends PrintWriter
synchronized (lock)
{
out.close();
+ _isClosed = true;
}
}
catch (IOException ex)
@@ -248,7 +257,7 @@ public class UncheckedPrintWriter extends PrintWriter
*/
@Override
public void write(char buf[])
- {
+ {
this.write(buf,0,buf.length);
}
diff --git a/jetty-io/src/main/java/org/eclipse/jetty/io/bio/SocketEndPoint.java b/jetty-io/src/main/java/org/eclipse/jetty/io/bio/SocketEndPoint.java
index f749f0ba417..67fb8ab214f 100644
--- a/jetty-io/src/main/java/org/eclipse/jetty/io/bio/SocketEndPoint.java
+++ b/jetty-io/src/main/java/org/eclipse/jetty/io/bio/SocketEndPoint.java
@@ -17,17 +17,13 @@ import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
+
import javax.net.ssl.SSLSocket;
import org.eclipse.jetty.util.StringUtil;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
-/**
- *
- * To change the template for this generated type comment go to
- * Window - Preferences - Java - Code Generation - Code and Comments
- */
public class SocketEndPoint extends StreamEndPoint
{
private static final Logger LOG = Log.getLogger(SocketEndPoint.class);
@@ -79,14 +75,34 @@ public class SocketEndPoint extends StreamEndPoint
@Override
public boolean isInputShutdown()
{
- return !isOpen() || super.isInputShutdown();
+ if (_socket instanceof SSLSocket)
+ return super.isInputShutdown();
+ return _socket.isClosed() || _socket.isInputShutdown();
}
/* ------------------------------------------------------------ */
@Override
public boolean isOutputShutdown()
{
- return !isOpen() || super.isOutputShutdown();
+ if (_socket instanceof SSLSocket)
+ return super.isOutputShutdown();
+
+ return _socket.isClosed() || _socket.isOutputShutdown();
+ }
+
+
+ /* ------------------------------------------------------------ */
+ /*
+ */
+ protected final void shutdownSocketOutput() throws IOException
+ {
+ if (!_socket.isClosed())
+ {
+ if (!_socket.isOutputShutdown())
+ _socket.shutdownOutput();
+ if (_socket.isInputShutdown())
+ _socket.close();
+ }
}
/* ------------------------------------------------------------ */
@@ -96,15 +112,27 @@ public class SocketEndPoint extends StreamEndPoint
@Override
public void shutdownOutput() throws IOException
{
- if (!isOutputShutdown())
- {
+ if (_socket instanceof SSLSocket)
super.shutdownOutput();
- if (!(_socket instanceof SSLSocket))
- _socket.shutdownOutput();
- }
+ else
+ shutdownSocketOutput();
}
+ /* ------------------------------------------------------------ */
+ /*
+ */
+ public void shutdownSocketInput() throws IOException
+ {
+ if (!_socket.isClosed())
+ {
+ if (!_socket.isInputShutdown())
+ _socket.shutdownInput();
+ if (_socket.isOutputShutdown())
+ _socket.close();
+ }
+ }
+
/* ------------------------------------------------------------ */
/*
* @see org.eclipse.jetty.io.bio.StreamEndPoint#shutdownOutput()
@@ -112,12 +140,10 @@ public class SocketEndPoint extends StreamEndPoint
@Override
public void shutdownInput() throws IOException
{
- if (!isInputShutdown())
- {
+ if (_socket instanceof SSLSocket)
super.shutdownInput();
- if (!(_socket instanceof SSLSocket))
- _socket.shutdownInput();
- }
+ else
+ shutdownSocketInput();
}
/* ------------------------------------------------------------ */
@@ -247,4 +273,9 @@ public class SocketEndPoint extends StreamEndPoint
}
}
+ @Override
+ public String toString()
+ {
+ return _local + " <--> " + _remote;
+ }
}
diff --git a/jetty-io/src/main/java/org/eclipse/jetty/io/bio/StreamEndPoint.java b/jetty-io/src/main/java/org/eclipse/jetty/io/bio/StreamEndPoint.java
index 48dce3723df..0ebab0058d3 100644
--- a/jetty-io/src/main/java/org/eclipse/jetty/io/bio/StreamEndPoint.java
+++ b/jetty-io/src/main/java/org/eclipse/jetty/io/bio/StreamEndPoint.java
@@ -72,10 +72,8 @@ public class StreamEndPoint implements EndPoint
public void shutdownOutput() throws IOException
{
- if (_oshut)
- return;
_oshut = true;
- if (_out!=null)
+ if (_ishut && _out!=null)
_out.close();
}
@@ -86,10 +84,8 @@ public class StreamEndPoint implements EndPoint
public void shutdownInput() throws IOException
{
- if (_ishut)
- return;
_ishut = true;
- if (_in!=null)
+ if (_oshut&&_in!=null)
_in.close();
}
@@ -122,6 +118,8 @@ public class StreamEndPoint implements EndPoint
*/
public int fill(Buffer buffer) throws IOException
{
+ if (_ishut)
+ return -1;
if (_in==null)
return 0;
@@ -135,15 +133,10 @@ public class StreamEndPoint implements EndPoint
try
{
- int read=buffer.readFrom(_in, space);
- if (read<0 && isOpen())
- {
- if (!isInputShutdown())
- shutdownInput();
- else if (isOutputShutdown())
- close();
- }
- return read;
+ int filled=buffer.readFrom(_in, space);
+ if (filled<0)
+ shutdownInput();
+ return filled;
}
catch(SocketTimeoutException e)
{
@@ -157,8 +150,10 @@ public class StreamEndPoint implements EndPoint
*/
public int flush(Buffer buffer) throws IOException
{
- if (_out==null)
+ if (_oshut)
return -1;
+ if (_out==null)
+ return 0;
int length=buffer.length();
if (length>0)
buffer.writeTo(_out);
@@ -310,24 +305,6 @@ public class StreamEndPoint implements EndPoint
_out.flush();
}
- /* ------------------------------------------------------------ */
- public boolean isBufferingInput()
- {
- return false;
- }
-
- /* ------------------------------------------------------------ */
- public boolean isBufferingOutput()
- {
- return false;
- }
-
- /* ------------------------------------------------------------ */
- public boolean isBufferred()
- {
- return false;
- }
-
/* ------------------------------------------------------------ */
public int getMaxIdleTime()
{
diff --git a/jetty-io/src/main/java/org/eclipse/jetty/io/bio/StringEndPoint.java b/jetty-io/src/main/java/org/eclipse/jetty/io/bio/StringEndPoint.java
index 1e8aad30a26..1698249d438 100644
--- a/jetty-io/src/main/java/org/eclipse/jetty/io/bio/StringEndPoint.java
+++ b/jetty-io/src/main/java/org/eclipse/jetty/io/bio/StringEndPoint.java
@@ -53,6 +53,8 @@ public class StringEndPoint extends StreamEndPoint
_in=_bin;
_bout = new ByteArrayOutputStream();
_out=_bout;
+ _ishut=false;
+ _oshut=false;
}
catch(Exception e)
{
diff --git a/jetty-io/src/main/java/org/eclipse/jetty/io/nio/AsyncConnection.java b/jetty-io/src/main/java/org/eclipse/jetty/io/nio/AsyncConnection.java
new file mode 100644
index 00000000000..717d01294ab
--- /dev/null
+++ b/jetty-io/src/main/java/org/eclipse/jetty/io/nio/AsyncConnection.java
@@ -0,0 +1,23 @@
+// ========================================================================
+// Copyright (c) 2010 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.io.nio;
+
+import java.io.IOException;
+
+import org.eclipse.jetty.io.Connection;
+
+public interface AsyncConnection extends Connection
+{
+ void onInputShutdown() throws IOException;
+}
diff --git a/jetty-io/src/main/java/org/eclipse/jetty/io/nio/ChannelEndPoint.java b/jetty-io/src/main/java/org/eclipse/jetty/io/nio/ChannelEndPoint.java
index 9aefa436cb9..fb3c09145a8 100644
--- a/jetty-io/src/main/java/org/eclipse/jetty/io/nio/ChannelEndPoint.java
+++ b/jetty-io/src/main/java/org/eclipse/jetty/io/nio/ChannelEndPoint.java
@@ -43,7 +43,9 @@ public class ChannelEndPoint implements EndPoint
protected final Socket _socket;
protected final InetSocketAddress _local;
protected final InetSocketAddress _remote;
- protected int _maxIdleTime;
+ protected volatile int _maxIdleTime;
+ private volatile boolean _ishut;
+ private volatile boolean _oshut;
public ChannelEndPoint(ByteChannel channel) throws IOException
{
@@ -102,20 +104,76 @@ public class ChannelEndPoint implements EndPoint
return _channel.isOpen();
}
+ /** Shutdown the channel Input.
+ * Cannot be overridden. To override, see {@link #shutdownInput()}
+ * @throws IOException
+ */
+ protected final void shutdownChannelInput() throws IOException
+ {
+ LOG.debug("ishut {}", this);
+ _ishut = true;
+ if (_channel.isOpen())
+ {
+ if (_socket != null)
+ {
+ try
+ {
+ if (!_socket.isInputShutdown())
+ {
+ _socket.shutdownInput();
+ }
+ }
+ catch (SocketException e)
+ {
+ LOG.debug(e.toString());
+ LOG.ignore(e);
+ }
+ finally
+ {
+ if (_oshut)
+ {
+ close();
+ }
+ }
+ }
+ }
+ }
+
/* (non-Javadoc)
* @see org.eclipse.io.EndPoint#close()
*/
public void shutdownInput() throws IOException
{
- if (_channel.isOpen() && _channel instanceof SocketChannel)
+ shutdownChannelInput();
+ }
+
+ protected final void shutdownChannelOutput() throws IOException
+ {
+ LOG.debug("oshut {}",this);
+ _oshut = true;
+ if (_channel.isOpen())
{
- Socket socket= ((SocketChannel)_channel).socket();
- if (!socket.isClosed())
+ if (_socket != null)
{
- if(socket.isOutputShutdown())
- socket.close();
- else if (!socket.isInputShutdown())
- socket.shutdownInput();
+ try
+ {
+ if (!_socket.isOutputShutdown())
+ {
+ _socket.shutdownOutput();
+ }
+ }
+ catch (SocketException e)
+ {
+ LOG.debug(e.toString());
+ LOG.ignore(e);
+ }
+ finally
+ {
+ if (_ishut)
+ {
+ close();
+ }
+ }
}
}
}
@@ -125,34 +183,17 @@ public class ChannelEndPoint implements EndPoint
*/
public void shutdownOutput() throws IOException
{
- if (_channel.isOpen() && _channel instanceof SocketChannel)
- {
- Socket socket= ((SocketChannel)_channel).socket();
- if (!socket.isClosed())
- {
- try
- {
- if (socket.isInputShutdown())
- socket.close();
- else if (!socket.isOutputShutdown())
- socket.shutdownOutput();
- }
- catch(SocketException e)
- {
- LOG.ignore(e);
- }
- }
- }
+ shutdownChannelOutput();
}
public boolean isOutputShutdown()
{
- return _channel.isOpen() && _socket!=null && _socket.isOutputShutdown();
+ return _oshut || !_channel.isOpen() || _socket != null && _socket.isOutputShutdown();
}
public boolean isInputShutdown()
{
- return _channel.isOpen() && _socket!=null && _socket.isInputShutdown();
+ return _ishut || !_channel.isOpen() || _socket != null && _socket.isInputShutdown();
}
/* (non-Javadoc)
@@ -160,6 +201,7 @@ public class ChannelEndPoint implements EndPoint
*/
public void close() throws IOException
{
+ LOG.debug("close {}",this);
_channel.close();
}
@@ -168,13 +210,15 @@ public class ChannelEndPoint implements EndPoint
*/
public int fill(Buffer buffer) throws IOException
{
+ if (_ishut)
+ return -1;
Buffer buf = buffer.buffer();
int len=0;
if (buf instanceof NIOBuffer)
{
final NIOBuffer nbuf = (NIOBuffer)buf;
final ByteBuffer bbuf=nbuf.getByteBuffer();
-
+
//noinspection SynchronizationOnLocalVariableOrMethodParameter
try
{
@@ -196,22 +240,24 @@ public class ChannelEndPoint implements EndPoint
{
if (!isInputShutdown())
shutdownInput();
- else if (isOutputShutdown())
+ if (isOutputShutdown())
_channel.close();
}
}
catch (IOException x)
{
- LOG.debug(x);
+ LOG.debug(x.toString());
+ LOG.ignore(x);
try
{
- close();
+ if (_channel.isOpen())
+ _channel.close();
}
- catch (IOException xx)
+ catch (Exception xx)
{
LOG.ignore(xx);
}
-
+
if (len>0)
throw x;
len=-1;
@@ -221,7 +267,7 @@ public class ChannelEndPoint implements EndPoint
{
throw new IOException("Not Implemented");
}
-
+
return len;
}
@@ -293,20 +339,6 @@ public class ChannelEndPoint implements EndPoint
}
else
{
- if (header!=null)
- {
- if (buffer!=null && buffer.length()>0 && header.space()>buffer.length())
- {
- header.put(buffer);
- buffer.clear();
- }
- if (trailer!=null && trailer.length()>0 && header.space()>trailer.length())
- {
- header.put(trailer);
- trailer.clear();
- }
- }
-
// flush header
if (header!=null && header.length()>0)
length=flush(header);
@@ -478,24 +510,6 @@ public class ChannelEndPoint implements EndPoint
{
}
- /* ------------------------------------------------------------ */
- public boolean isBufferingInput()
- {
- return false;
- }
-
- /* ------------------------------------------------------------ */
- public boolean isBufferingOutput()
- {
- return false;
- }
-
- /* ------------------------------------------------------------ */
- public boolean isBufferred()
- {
- return false;
- }
-
/* ------------------------------------------------------------ */
public int getMaxIdleTime()
{
diff --git a/jetty-io/src/main/java/org/eclipse/jetty/io/nio/SelectChannelEndPoint.java b/jetty-io/src/main/java/org/eclipse/jetty/io/nio/SelectChannelEndPoint.java
index a318543f647..39cd2205dc3 100644
--- a/jetty-io/src/main/java/org/eclipse/jetty/io/nio/SelectChannelEndPoint.java
+++ b/jetty-io/src/main/java/org/eclipse/jetty/io/nio/SelectChannelEndPoint.java
@@ -27,6 +27,7 @@ import org.eclipse.jetty.io.EofException;
import org.eclipse.jetty.io.nio.SelectorManager.SelectSet;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
+import org.eclipse.jetty.util.thread.Timeout.Task;
/* ------------------------------------------------------------ */
/**
@@ -35,7 +36,7 @@ import org.eclipse.jetty.util.log.Logger;
public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPoint, ConnectedEndPoint
{
public static final Logger LOG=Log.getLogger("org.eclipse.jetty.io.nio");
-
+
private final SelectorManager.SelectSet _selectSet;
private final SelectorManager _manager;
private SelectionKey _key;
@@ -46,36 +47,38 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
/** The desired value for {@link SelectionKey#interestOps()} */
private int _interestOps;
-
+
/**
* The connection instance is the handler for any IO activity on the endpoint.
- * There is a different type of connection for HTTP, AJP, WebSocket and
- * ProxyConnect. The connection may change for an SCEP as it is upgraded
+ * There is a different type of connection for HTTP, AJP, WebSocket and
+ * ProxyConnect. The connection may change for an SCEP as it is upgraded
* from HTTP to proxy connect or websocket.
*/
- private volatile Connection _connection;
-
+ private volatile AsyncConnection _connection;
+
/** true if a thread has been dispatched to handle this endpoint */
private boolean _dispatched = false;
-
+
/** true if a non IO dispatch (eg async resume) is outstanding */
- private boolean _redispatched = false;
-
+ private boolean _asyncDispatch = false;
+
/** true if the last write operation succeed and wrote all offered bytes */
private volatile boolean _writable = true;
-
+
/** True if a thread has is blocked in {@link #blockReadable(long)} */
private boolean _readBlocked;
/** True if a thread has is blocked in {@link #blockWritable(long)} */
private boolean _writeBlocked;
-
+
/** true if {@link SelectSet#destroyEndPoint(SelectChannelEndPoint)} has not been called */
private boolean _open;
-
+
private volatile long _idleTimestamp;
-
+
+ private boolean _ishut;
+
/* ------------------------------------------------------------ */
public SelectChannelEndPoint(SocketChannel channel, SelectSet selectSet, SelectionKey key, int maxIdleTime)
throws IOException
@@ -85,33 +88,13 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
_manager = selectSet.getManager();
_selectSet = selectSet;
_dispatched = false;
- _redispatched = false;
+ _asyncDispatch = false;
_open=true;
_key = key;
- _connection = _manager.newConnection(channel,this);
-
- scheduleIdle();
+ setCheckForIdle(true);
}
- /* ------------------------------------------------------------ */
- public SelectChannelEndPoint(SocketChannel channel, SelectSet selectSet, SelectionKey key)
- throws IOException
- {
- super(channel);
-
- _manager = selectSet.getManager();
- _selectSet = selectSet;
- _dispatched = false;
- _redispatched = false;
- _open=true;
- _key = key;
-
- _connection = _manager.newConnection(channel,this);
-
- scheduleIdle();
- }
-
/* ------------------------------------------------------------ */
public SelectionKey getSelectionKey()
{
@@ -137,8 +120,9 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
public void setConnection(Connection connection)
{
Connection old=_connection;
- _connection=connection;
- _manager.endPointUpgraded(this,old);
+ _connection=(AsyncConnection)connection;
+ if (old!=null && old!=_connection)
+ _manager.endPointUpgraded(this,old);
}
/* ------------------------------------------------------------ */
@@ -146,7 +130,7 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
{
return _idleTimestamp;
}
-
+
/* ------------------------------------------------------------ */
/** Called by selectSet to schedule handling
*
@@ -176,17 +160,10 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
// wake them up is as good as a dispatched.
this.notifyAll();
- // we are not interested in further selecting
- if (_dispatched)
- _key.interestOps(0);
- return;
- }
-
- // Otherwise if we are still dispatched
- if (!isReadyForDispatch())
- {
// we are not interested in further selecting
_key.interestOps(0);
+ if (!_dispatched)
+ updateKey();
return;
}
@@ -211,6 +188,18 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
}
}
+ /* ------------------------------------------------------------ */
+ public void asyncDispatch()
+ {
+ synchronized(this)
+ {
+ if (_dispatched)
+ _asyncDispatch=true;
+ else
+ dispatch();
+ }
+ }
+
/* ------------------------------------------------------------ */
public void dispatch()
{
@@ -218,7 +207,7 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
{
if (_dispatched)
{
- _redispatched=true;
+ throw new IllegalStateException("dispatched");
}
else
{
@@ -245,9 +234,9 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
{
synchronized (this)
{
- if (_redispatched)
+ if (_asyncDispatch)
{
- _redispatched=false;
+ _asyncDispatch=false;
return false;
}
_dispatched = false;
@@ -257,48 +246,74 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
}
/* ------------------------------------------------------------ */
- public void scheduleIdle()
+ public void cancelTimeout(Task task)
{
- _idleTimestamp=System.currentTimeMillis();
+ getSelectSet().cancelTimeout(task);
}
/* ------------------------------------------------------------ */
- public void cancelIdle()
+ public void scheduleTimeout(Task task, long timeoutMs)
{
- _idleTimestamp=0;
+ getSelectSet().scheduleTimeout(task,timeoutMs);
+ }
+
+ /* ------------------------------------------------------------ */
+ public void setCheckForIdle(boolean check)
+ {
+ _idleTimestamp=check?System.currentTimeMillis():0;
+ }
+
+ /* ------------------------------------------------------------ */
+ public boolean isCheckForIdle()
+ {
+ return _idleTimestamp!=0;
+ }
+
+ /* ------------------------------------------------------------ */
+ protected void notIdle()
+ {
+ if (_idleTimestamp!=0)
+ _idleTimestamp=System.currentTimeMillis();
}
/* ------------------------------------------------------------ */
public void checkIdleTimestamp(long now)
{
long idleTimestamp=_idleTimestamp;
- if (!getChannel().isOpen() || idleTimestamp!=0 && _maxIdleTime>0 && now>(idleTimestamp+_maxIdleTime))
- idleExpired();
+
+ if (idleTimestamp!=0 && _maxIdleTime>0)
+ {
+ long idleForMs=now-idleTimestamp;
+ if (idleForMs>_maxIdleTime)
+ {
+ onIdleExpired(idleForMs);
+ _idleTimestamp=now;
+ }
+ }
}
/* ------------------------------------------------------------ */
- protected void idleExpired()
+ public void onIdleExpired(long idleForMs)
{
- _connection.idleExpired();
+ _connection.onIdleExpired(idleForMs);
}
/* ------------------------------------------------------------ */
- /**
- * @return True if the endpoint has produced/consumed bytes itself (non application data).
- */
- public boolean isProgressing()
+ @Override
+ public int fill(Buffer buffer) throws IOException
{
- return false;
+ int fill=super.fill(buffer);
+ if (fill>0)
+ notIdle();
+ return fill;
}
-
+
/* ------------------------------------------------------------ */
- /*
- */
@Override
public int flush(Buffer header, Buffer buffer, Buffer trailer) throws IOException
{
int l = super.flush(header, buffer, trailer);
-
+
// If there was something to write and it wasn't written, then we are not writable.
if (l==0 && ( header!=null && header.hasContent() || buffer!=null && buffer.hasContent() || trailer!=null && trailer.hasContent()))
{
@@ -309,8 +324,11 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
updateKey();
}
}
- else
+ else if (l>0)
+ {
_writable=true;
+ notIdle();
+ }
return l;
}
@@ -321,7 +339,7 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
public int flush(Buffer buffer) throws IOException
{
int l = super.flush(buffer);
-
+
// If there was something to write and it wasn't written, then we are not writable.
if (l==0 && buffer!=null && buffer.hasContent())
{
@@ -332,20 +350,13 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
updateKey();
}
}
- else
- _writable=true;
-
- return l;
- }
-
- /* ------------------------------------------------------------ */
- public boolean isReadyForDispatch()
- {
- synchronized (this)
+ else if (l>0)
{
- // Ready if not dispatched and not suspended
- return !(_dispatched || getConnection().isSuspended());
+ _writable=true;
+ notIdle();
}
+
+ return l;
}
/* ------------------------------------------------------------ */
@@ -357,12 +368,17 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
{
synchronized (this)
{
+ if (isInputShutdown())
+ throw new EofException();
+
long now=_selectSet.getNow();
long end=now+timeoutMs;
+ boolean check=isCheckForIdle();
+ setCheckForIdle(true);
try
{
_readBlocked=true;
- while (isOpen() && _readBlocked)
+ while (!isInputShutdown() && _readBlocked)
{
try
{
@@ -385,6 +401,7 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
finally
{
_readBlocked=false;
+ setCheckForIdle(check);
}
}
return true;
@@ -399,15 +416,17 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
{
synchronized (this)
{
- if (!isOpen() || isOutputShutdown())
+ if (isOutputShutdown())
throw new EofException();
-
+
long now=_selectSet.getNow();
long end=now+timeoutMs;
+ boolean check=isCheckForIdle();
+ setCheckForIdle(true);
try
{
_writeBlocked=true;
- while (isOpen() && _writeBlocked && !isOutputShutdown())
+ while (_writeBlocked && !isOutputShutdown())
{
try
{
@@ -426,21 +445,10 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
return false;
}
}
- catch(Throwable e)
- {
- // TODO remove this if it finds nothing
- LOG.warn(e);
- if (e instanceof RuntimeException)
- throw (RuntimeException)e;
- if (e instanceof Error)
- throw (Error)e;
- throw new RuntimeException(e);
- }
finally
{
_writeBlocked=false;
- if (_idleTimestamp!=-1)
- scheduleIdle();
+ setCheckForIdle(check);
}
}
return true;
@@ -452,17 +460,32 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
{
_writable=false;
}
-
+
/* ------------------------------------------------------------ */
+ /**
+ * @see org.eclipse.jetty.io.AsyncEndPoint#scheduleWrite()
+ */
public void scheduleWrite()
{
if (_writable==true)
LOG.debug("Required scheduleWrite {}",this);
-
+
_writable=false;
updateKey();
}
+ /* ------------------------------------------------------------ */
+ public boolean isWritable()
+ {
+ return _writable;
+ }
+
+ /* ------------------------------------------------------------ */
+ public boolean hasProgressed()
+ {
+ return false;
+ }
+
/* ------------------------------------------------------------ */
/**
* Updates selection key. Adds operations types to the selection key as needed. No operations
@@ -471,17 +494,21 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
*/
private void updateKey()
{
+ final boolean changed;
synchronized (this)
{
- int ops=-1;
+ int current_ops=-1;
if (getChannel().isOpen())
{
+ boolean read_interest = _readBlocked || (!_dispatched && !_connection.isSuspended());
+ boolean write_interest= _writeBlocked || (!_dispatched && !_writable);
+
_interestOps =
- ((!_socket.isInputShutdown() && (!_dispatched || _readBlocked)) ? SelectionKey.OP_READ : 0)
- | ((!_socket.isOutputShutdown()&& (!_writable || _writeBlocked)) ? SelectionKey.OP_WRITE : 0);
+ ((!_socket.isInputShutdown() && read_interest ) ? SelectionKey.OP_READ : 0)
+ | ((!_socket.isOutputShutdown()&& write_interest) ? SelectionKey.OP_WRITE : 0);
try
{
- ops = ((_key!=null && _key.isValid())?_key.interestOps():-1);
+ current_ops = ((_key!=null && _key.isValid())?_key.interestOps():-1);
}
catch(Exception e)
{
@@ -489,14 +516,17 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
LOG.ignore(e);
}
}
-
- if(_interestOps == ops && getChannel().isOpen())
- return;
+ changed=_interestOps!=current_ops;
+ }
+
+ if(changed)
+ {
+ _selectSet.addChange(this);
+ _selectSet.wakeup();
}
- _selectSet.addChange(this);
- _selectSet.wakeup();
}
+
/* ------------------------------------------------------------ */
/**
* Synchronize the interestOps with the actual key. Call is scheduled by a call to updateKey
@@ -529,7 +559,6 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
{
_key.cancel();
}
- cancelIdle();
if (_open)
{
@@ -558,7 +587,6 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
if (_key!=null && _key.isValid())
_key.cancel();
- cancelIdle();
if (_open)
{
_open=false;
@@ -583,11 +611,13 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
{
while(true)
{
- final Connection next = _connection.handle();
+ final AsyncConnection next = (AsyncConnection)_connection.handle();
if (next!=_connection)
{
LOG.debug("{} replaced {}",next,_connection);
+ Connection old=_connection;
_connection=next;
+ _manager.endPointUpgraded(this,old);
continue;
}
break;
@@ -600,23 +630,44 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
catch (EofException e)
{
LOG.debug("EOF", e);
- try{getChannel().close();}
+ try{close();}
catch(IOException e2){LOG.ignore(e2);}
}
catch (IOException e)
{
LOG.warn(e.toString());
LOG.debug(e);
- try{getChannel().close();}
+ try{close();}
catch(IOException e2){LOG.ignore(e2);}
}
catch (Throwable e)
{
LOG.warn("handle failed", e);
- try{getChannel().close();}
+ try{close();}
catch(IOException e2){LOG.ignore(e2);}
}
- dispatched=!undispatch();
+ finally
+ {
+ if (!_ishut && isInputShutdown() && isOpen())
+ {
+ _ishut=true;
+ try
+ {
+ _connection.onInputShutdown();
+ }
+ catch(Throwable x)
+ {
+ LOG.warn("onInputShutdown failed", x);
+ try{close();}
+ catch(IOException e2){LOG.ignore(e2);}
+ }
+ finally
+ {
+ updateKey();
+ }
+ }
+ dispatched=!undispatch();
+ }
}
}
finally
@@ -660,9 +711,21 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
{
synchronized(this)
{
- return "SCEP@" + hashCode() + _channel+
- "[o="+isOpen()+" d=" + _dispatched + ",io=" + _interestOps+
- ",w=" + _writable + ",rb=" + _readBlocked + ",wb=" + _writeBlocked + "]";
+ return String.format("SCEP@%x{%s->%s,d=%b,open=%b,ishut=%b,oshut=%b,rb=%b,wb=%b,w=%b,i=%d%s%s%s}",
+ hashCode(),
+ _socket.getRemoteSocketAddress(),
+ _socket.getLocalSocketAddress(),
+ _dispatched,
+ isOpen(),
+ isInputShutdown(),
+ isOutputShutdown(),
+ _readBlocked,
+ _writeBlocked,
+ _writable,
+ _interestOps,
+ _key != null && _key.isValid() ? "" : "!",
+ _key != null && _key.isValid() && _key.isReadable() ? "r" : "",
+ _key != null && _key.isValid() && _key.isWritable() ? "w" : "");
}
}
diff --git a/jetty-io/src/main/java/org/eclipse/jetty/io/nio/SelectorManager.java b/jetty-io/src/main/java/org/eclipse/jetty/io/nio/SelectorManager.java
index 1b8218832a4..eaeabc88624 100644
--- a/jetty-io/src/main/java/org/eclipse/jetty/io/nio/SelectorManager.java
+++ b/jetty-io/src/main/java/org/eclipse/jetty/io/nio/SelectorManager.java
@@ -30,6 +30,7 @@ import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import org.eclipse.jetty.io.AsyncEndPoint;
import org.eclipse.jetty.io.ConnectedEndPoint;
import org.eclipse.jetty.io.Connection;
import org.eclipse.jetty.io.EndPoint;
@@ -54,7 +55,7 @@ public abstract class SelectorManager extends AbstractLifeCycle implements Dumpa
public static final Logger LOG=Log.getLogger("org.eclipse.jetty.io.nio");
private static final int __MONITOR_PERIOD=Integer.getInteger("org.eclipse.jetty.io.nio.MONITOR_PERIOD",1000).intValue();
- private static final int __MAX_SELECTS=Integer.getInteger("org.eclipse.jetty.io.nio.MAX_SELECTS",25000).intValue();
+ private static final int __MAX_SELECTS=Integer.getInteger("org.eclipse.jetty.io.nio.MAX_SELECTS",100000).intValue();
private static final int __BUSY_PAUSE=Integer.getInteger("org.eclipse.jetty.io.nio.BUSY_PAUSE",50).intValue();
private static final int __IDLE_TICK=Integer.getInteger("org.eclipse.jetty.io.nio.IDLE_TICK",400).intValue();
@@ -63,7 +64,7 @@ public abstract class SelectorManager extends AbstractLifeCycle implements Dumpa
private long _lowResourcesConnections;
private SelectSet[] _selectSet;
private int _selectSets=1;
- private volatile int _set;
+ private volatile int _set=0;
private boolean _deferringInterestedOps0=true;
private int _selectorPriorityDelta=0;
@@ -128,6 +129,8 @@ public abstract class SelectorManager extends AbstractLifeCycle implements Dumpa
// be distributed over the available sets.
int s=_set++;
+ if (s<0)
+ s=-s;
s=s%_selectSets;
SelectSet[] sets=_selectSet;
if (sets!=null)
@@ -150,6 +153,8 @@ public abstract class SelectorManager extends AbstractLifeCycle implements Dumpa
// be distributed over the available sets.
int s=_set++;
+ if (s<0)
+ s=-s;
s=s%_selectSets;
SelectSet[] sets=_selectSet;
if (sets!=null)
@@ -167,6 +172,8 @@ public abstract class SelectorManager extends AbstractLifeCycle implements Dumpa
public void register(ServerSocketChannel acceptChannel)
{
int s=_set++;
+ if (s<0)
+ s=-s;
s=s%_selectSets;
SelectSet set=_selectSet[s];
set.addChange(acceptChannel);
@@ -276,10 +283,6 @@ public abstract class SelectorManager extends AbstractLifeCycle implements Dumpa
{
set.doSelect();
}
- catch(ThreadDeath e)
- {
- throw e;
- }
catch(IOException e)
{
LOG.ignore(e);
@@ -337,7 +340,7 @@ public abstract class SelectorManager extends AbstractLifeCycle implements Dumpa
protected abstract void endPointUpgraded(ConnectedEndPoint endpoint,Connection oldConnection);
/* ------------------------------------------------------------------------------- */
- protected abstract Connection newConnection(SocketChannel channel, SelectChannelEndPoint endpoint);
+ public abstract AsyncConnection newConnection(SocketChannel channel, AsyncEndPoint endpoint, Object attachment);
/* ------------------------------------------------------------ */
/**
@@ -502,9 +505,6 @@ public abstract class SelectorManager extends AbstractLifeCycle implements Dumpa
}
catch (Throwable e)
{
- if (e instanceof ThreadDeath)
- throw (ThreadDeath)e;
-
if (isRunning())
LOG.warn(e);
else
@@ -570,21 +570,13 @@ public abstract class SelectorManager extends AbstractLifeCycle implements Dumpa
{
// Start injecting pauses
_pausing=true;
-
+
// if this is the first pause
if (!_paused)
{
// Log and dump some status
_paused=true;
LOG.warn("Selector {} is too busy, pausing!",this);
- final SelectSet set = this;
- SelectorManager.this.dispatch(
- new Runnable(){
- public void run()
- {
- System.err.println(set+":\n"+set.dump());
- }
- });
}
}
}
@@ -713,17 +705,18 @@ public abstract class SelectorManager extends AbstractLifeCycle implements Dumpa
endp.checkIdleTimestamp(idle_now);
}
}
+ public String toString() {return "Idle-"+super.toString();}
});
-
+
}
-
+
// Reset busy select monitor counts
if (__MONITOR_PERIOD>0 && now>_monitorNext)
{
_busySelects=0;
_pausing=false;
_monitorNext=now+__MONITOR_PERIOD;
-
+
}
}
catch (ClosedSelectorException e)
@@ -837,6 +830,7 @@ public abstract class SelectorManager extends AbstractLifeCycle implements Dumpa
private SelectChannelEndPoint createEndPoint(SocketChannel channel, SelectionKey sKey) throws IOException
{
SelectChannelEndPoint endp = newEndPoint(channel,this,sKey);
+ LOG.debug("created {}",endp);
endPointOpened(endp);
_endPoints.put(endp,this);
return endp;
@@ -975,11 +969,22 @@ public abstract class SelectorManager extends AbstractLifeCycle implements Dumpa
for (SelectionKey key: selector.keys())
{
if (key.isValid())
- dumpto.add(key.attachment()+" "+key.interestOps()+" "+key.readyOps());
+ dumpto.add(key.attachment()+" iOps="+key.interestOps()+" rOps="+key.readyOps());
else
- dumpto.add(key.attachment()+" - - ");
+ dumpto.add(key.attachment()+" iOps=-1 rOps=-1");
}
}
+
+ /* ------------------------------------------------------------ */
+ public String toString()
+ {
+ Selector selector=_selector;
+ return String.format("%s %s keys=%d selected=%d",
+ super.toString(),
+ SelectorManager.this.getState(),
+ selector != null && selector.isOpen() ? selector.keys().size() : -1,
+ selector != null && selector.isOpen() ? selector.selectedKeys().size() : -1);
+ }
}
/* ------------------------------------------------------------ */
diff --git a/jetty-io/src/main/java/org/eclipse/jetty/io/nio/SslConnection.java b/jetty-io/src/main/java/org/eclipse/jetty/io/nio/SslConnection.java
new file mode 100644
index 00000000000..d8cee6277f5
--- /dev/null
+++ b/jetty-io/src/main/java/org/eclipse/jetty/io/nio/SslConnection.java
@@ -0,0 +1,812 @@
+// ========================================================================
+// Copyright (c) 2004-2011 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.io.nio;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.concurrent.atomic.AtomicBoolean;
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLEngineResult;
+import javax.net.ssl.SSLEngineResult.HandshakeStatus;
+import javax.net.ssl.SSLException;
+import javax.net.ssl.SSLSession;
+
+import org.eclipse.jetty.io.AbstractConnection;
+import org.eclipse.jetty.io.AsyncEndPoint;
+import org.eclipse.jetty.io.Buffer;
+import org.eclipse.jetty.io.Connection;
+import org.eclipse.jetty.io.EndPoint;
+import org.eclipse.jetty.util.log.Log;
+import org.eclipse.jetty.util.log.Logger;
+import org.eclipse.jetty.util.thread.Timeout.Task;
+
+/* ------------------------------------------------------------ */
+/** SSL Connection.
+ * An AysyncConnection that acts as an interceptor between and EndPoint and another
+ * Connection, that implements TLS encryption using an {@link SSLEngine}.
+ *
+ * The connector uses an {@link AsyncEndPoint} (like {@link SelectChannelEndPoint}) as
+ * it's source/sink of encrypted data. It then provides {@link #getSslEndPoint()} to
+ * expose a source/sink of unencrypted data to another connection (eg HttpConnection).
+ */
+public class SslConnection extends AbstractConnection implements AsyncConnection
+{
+ static final Logger LOG=Log.getLogger("org.eclipse.jetty.io.nio.ssl");
+
+ private static final NIOBuffer __ZERO_BUFFER=new IndirectNIOBuffer(0);
+
+ private static final ThreadLocal __buffers = new ThreadLocal();
+ private final SSLEngine _engine;
+ private final SSLSession _session;
+ private AsyncConnection _connection;
+ private final SslEndPoint _sslEndPoint = new SslEndPoint();
+ private int _allocations;
+ private SslBuffers _buffers;
+ private NIOBuffer _inbound;
+ private NIOBuffer _unwrapBuf;
+ private NIOBuffer _outbound;
+ private AsyncEndPoint _aEndp;
+ private boolean _allowRenegotiate=true;
+ private boolean _handshook;
+ private boolean _ishut;
+ private boolean _oshut;
+ private final AtomicBoolean _progressed = new AtomicBoolean();
+
+ /* ------------------------------------------------------------ */
+ /* this is a half baked buffer pool
+ */
+ private static class SslBuffers
+ {
+ final NIOBuffer _in;
+ final NIOBuffer _out;
+ final NIOBuffer _unwrap;
+
+ SslBuffers(int packetSize, int appSize)
+ {
+ _in=new IndirectNIOBuffer(packetSize);
+ _out=new IndirectNIOBuffer(packetSize);
+ _unwrap=new IndirectNIOBuffer(appSize);
+ }
+ }
+
+ /* ------------------------------------------------------------ */
+ public SslConnection(SSLEngine engine,EndPoint endp)
+ {
+ this(engine,endp,System.currentTimeMillis());
+ }
+
+ /* ------------------------------------------------------------ */
+ public SslConnection(SSLEngine engine,EndPoint endp, long timeStamp)
+ {
+ super(endp,timeStamp);
+ _engine=engine;
+ _session=_engine.getSession();
+ _aEndp=(AsyncEndPoint)endp;
+ }
+
+ /* ------------------------------------------------------------ */
+ /**
+ * @return True if SSL re-negotiation is allowed (default false)
+ */
+ public boolean isAllowRenegotiate()
+ {
+ return _allowRenegotiate;
+ }
+
+ /* ------------------------------------------------------------ */
+ /**
+ * Set if SSL re-negotiation is allowed. CVE-2009-3555 discovered
+ * a vulnerability in SSL/TLS with re-negotiation. If your JVM
+ * does not have CVE-2009-3555 fixed, then re-negotiation should
+ * not be allowed. CVE-2009-3555 was fixed in Sun java 1.6 with a ban
+ * of renegotiates in u19 and with RFC5746 in u22.
+ *
+ * @param allowRenegotiate
+ * true if re-negotiation is allowed (default false)
+ */
+ public void setAllowRenegotiate(boolean allowRenegotiate)
+ {
+ _allowRenegotiate = allowRenegotiate;
+ }
+
+ /* ------------------------------------------------------------ */
+ private void allocateBuffers()
+ {
+ synchronized (this)
+ {
+ if (_allocations++==0)
+ {
+ if (_buffers==null)
+ {
+ _buffers=__buffers.get();
+ if (_buffers==null)
+ _buffers=new SslBuffers(_session.getPacketBufferSize()*2,_session.getApplicationBufferSize()*2);
+ _inbound=_buffers._in;
+ _outbound=_buffers._out;
+ _unwrapBuf=_buffers._unwrap;
+ __buffers.set(null);
+ }
+ }
+ }
+ }
+
+ /* ------------------------------------------------------------ */
+ private void releaseBuffers()
+ {
+ synchronized (this)
+ {
+ if (--_allocations==0)
+ {
+ if (_buffers!=null &&
+ _inbound.length()==0 &&
+ _outbound.length()==0 &&
+ _unwrapBuf.length()==0)
+ {
+ _inbound=null;
+ _outbound=null;
+ _unwrapBuf=null;
+ __buffers.set(_buffers);
+ _buffers=null;
+ }
+ }
+ }
+ }
+
+ /* ------------------------------------------------------------ */
+ public Connection handle() throws IOException
+ {
+ try
+ {
+ allocateBuffers();
+
+ boolean progress=true;
+
+ while (progress)
+ {
+ progress=false;
+
+ // If we are handshook let the delegate connection
+ if (_engine.getHandshakeStatus()!=HandshakeStatus.NOT_HANDSHAKING)
+ {
+ progress=process(null,null);
+ }
+ else
+ {
+ // handle the delegate connection
+ AsyncConnection next = (AsyncConnection)_connection.handle();
+ if (next!=_connection && next!=null)
+ {
+ _connection=next;
+ progress=true;
+ }
+ }
+
+ LOG.debug("{} handle {} progress=",_session,this, progress);
+ }
+ }
+ finally
+ {
+ releaseBuffers();
+
+ if (!_ishut && _sslEndPoint.isInputShutdown() && _sslEndPoint.isOpen())
+ {
+ _ishut=true;
+ try
+ {
+ _connection.onInputShutdown();
+ }
+ catch(Throwable x)
+ {
+ LOG.warn("onInputShutdown failed", x);
+ try{_sslEndPoint.close();}
+ catch(IOException e2){LOG.ignore(e2);}
+ }
+ }
+ }
+
+ return this;
+ }
+
+ /* ------------------------------------------------------------ */
+ public boolean isIdle()
+ {
+ return false;
+ }
+
+ /* ------------------------------------------------------------ */
+ public boolean isSuspended()
+ {
+ return false;
+ }
+
+ /* ------------------------------------------------------------ */
+ public void onClose()
+ {
+ }
+
+ /* ------------------------------------------------------------ */
+ @Override
+ public void onIdleExpired(long idleForMs)
+ {
+ try
+ {
+ LOG.debug("onIdleExpired {}ms on {}",idleForMs,this);
+ _sslEndPoint.shutdownOutput();
+ }
+ catch (IOException e)
+ {
+ LOG.warn(e);
+ super.onIdleExpired(idleForMs);
+ }
+ }
+
+ /* ------------------------------------------------------------ */
+ public void onInputShutdown() throws IOException
+ {
+
+ }
+
+ /* ------------------------------------------------------------ */
+ private synchronized boolean process(Buffer toFill, Buffer toFlush) throws IOException
+ {
+ boolean some_progress=false;
+ try
+ {
+ allocateBuffers();
+ if (toFill==null)
+ {
+ _unwrapBuf.compact();
+ toFill=_unwrapBuf;
+ }
+ else if (toFill.capacity()<_session.getApplicationBufferSize())
+ {
+ boolean progress=process(null,toFlush);
+ if (_unwrapBuf!=null && _unwrapBuf.hasContent())
+ {
+ _unwrapBuf.skip(toFill.put(_unwrapBuf));
+ return true;
+ }
+ else
+ return progress;
+ }
+ else if (_unwrapBuf!=null && _unwrapBuf.hasContent())
+ {
+ _unwrapBuf.skip(toFill.put(_unwrapBuf));
+ return true;
+ }
+
+
+ if (toFlush==null)
+ toFlush=__ZERO_BUFFER;
+
+ boolean progress=true;
+
+ while (progress)
+ {
+ progress=false;
+ int filled=0,flushed=0;
+
+ try
+ {
+ // Read any available data
+ if (_inbound.space()>0 && (filled=_endp.fill(_inbound))>0)
+ progress = true;
+
+ // flush any output data
+ if (_outbound.hasContent() && (flushed=_endp.flush(_outbound))>0)
+ progress = true;
+ }
+ catch (Exception e)
+ {
+ LOG.debug(e.toString());
+ LOG.ignore(e);
+ }
+ LOG.debug("{} {} {} filled={}/{} flushed={}/{}",_session,this,_engine.getHandshakeStatus(),filled,_inbound.length(),flushed,_outbound.length());
+
+ // handle the current hand share status
+ switch(_engine.getHandshakeStatus())
+ {
+ case FINISHED:
+ throw new IllegalStateException();
+
+ case NOT_HANDSHAKING:
+ {
+ // Try wrapping some application data
+ if (toFlush.hasContent() && _outbound.space()>0 && wrap(toFlush))
+ progress=true;
+
+ // Try unwrapping some application data
+ if (toFill.space()>0 && _inbound.hasContent() && unwrap(toFill))
+ progress=true;
+ }
+ break;
+
+ case NEED_TASK:
+ {
+ // A task needs to be run, so run it!
+ Runnable task;
+ while ((task=_engine.getDelegatedTask())!=null)
+ {
+ progress=true;
+ task.run();
+ }
+
+ // Detect SUN JVM Bug!!!
+ /* TODO
+ if(initialStatus==HandshakeStatus.NOT_HANDSHAKING &&
+ _engine.getHandshakeStatus()==HandshakeStatus.NEED_UNWRAP && sent==0 )
+ {
+ // This should be NEED_WRAP
+ // The fix simply detects the signature of the bug and then close the connection (fail-fast) so that ff3 will delegate to using SSL instead of TLS.
+ // This is a jvm bug on java1.6 where the SSLEngine expects more data from the initial handshake when the client(ff3-tls) already had given it.
+ // See http://jira.codehaus.org/browse/JETTY-567 for more details
+ LOG.warn("{} JETTY-567",_session);
+ _endp.close();
+ return false;
+ }
+ */
+ }
+ break;
+
+ case NEED_WRAP:
+ {
+ // The SSL needs to send some handshake data to the other side
+ if (_handshook && !_allowRenegotiate)
+ _endp.close();
+ else if (wrap(toFlush))
+ progress=true;
+ }
+ break;
+
+ case NEED_UNWRAP:
+ {
+ // The SSL needs to receive some handshake data from the other side
+ if (_handshook && !_allowRenegotiate)
+ _endp.close();
+ else if (unwrap(toFill))
+ progress=true;
+ }
+ break;
+ }
+
+ // pass on ishut/oshut state
+ if (_endp.isOpen() && _endp.isInputShutdown() && !_inbound.hasContent())
+ _engine.closeInbound();
+
+ if (_endp.isOpen() && _engine.isOutboundDone() && !_outbound.hasContent())
+ _endp.shutdownOutput();
+
+ some_progress|=progress;
+ }
+ }
+ finally
+ {
+ releaseBuffers();
+ if (some_progress)
+ _progressed.set(true);
+ }
+ return some_progress;
+ }
+
+ private synchronized boolean wrap(final Buffer buffer) throws IOException
+ {
+ ByteBuffer bbuf=extractByteBuffer(buffer);
+ final SSLEngineResult result;
+
+ synchronized(bbuf)
+ {
+ _outbound.compact();
+ ByteBuffer out_buffer=_outbound.getByteBuffer();
+ synchronized(out_buffer)
+ {
+ try
+ {
+ bbuf.position(buffer.getIndex());
+ bbuf.limit(buffer.putIndex());
+ out_buffer.position(_outbound.putIndex());
+ out_buffer.limit(out_buffer.capacity());
+ result=_engine.wrap(bbuf,out_buffer);
+ if (LOG.isDebugEnabled())
+ LOG.debug("{} wrap {} {} consumed={} produced={}",
+ _session,
+ result.getStatus(),
+ result.getHandshakeStatus(),
+ result.bytesConsumed(),
+ result.bytesProduced());
+
+
+ buffer.skip(result.bytesConsumed());
+ _outbound.setPutIndex(_outbound.putIndex()+result.bytesProduced());
+ }
+ catch(SSLException e)
+ {
+ LOG.warn(_endp+":",e);
+ _endp.close();
+ throw e;
+ }
+ finally
+ {
+ out_buffer.position(0);
+ out_buffer.limit(out_buffer.capacity());
+ bbuf.position(0);
+ bbuf.limit(bbuf.capacity());
+ }
+ }
+ }
+
+ switch(result.getStatus())
+ {
+ case BUFFER_UNDERFLOW:
+ throw new IllegalStateException();
+
+ case BUFFER_OVERFLOW:
+ break;
+
+ case OK:
+ if (result.getHandshakeStatus()==HandshakeStatus.FINISHED)
+ _handshook=true;
+ break;
+
+ case CLOSED:
+ LOG.debug("wrap CLOSE {} {}",this,result);
+ if (result.getHandshakeStatus()==HandshakeStatus.FINISHED)
+ _endp.close();
+ break;
+
+ default:
+ LOG.warn("{} wrap default {}",_session,result);
+ throw new IOException(result.toString());
+ }
+
+ return result.bytesConsumed()>0 || result.bytesProduced()>0;
+ }
+
+ private synchronized boolean unwrap(final Buffer buffer) throws IOException
+ {
+ if (!_inbound.hasContent())
+ return false;
+
+ ByteBuffer bbuf=extractByteBuffer(buffer);
+ final SSLEngineResult result;
+
+ synchronized(bbuf)
+ {
+ ByteBuffer in_buffer=_inbound.getByteBuffer();
+ synchronized(in_buffer)
+ {
+ try
+ {
+ bbuf.position(buffer.putIndex());
+ bbuf.limit(buffer.capacity());
+ in_buffer.position(_inbound.getIndex());
+ in_buffer.limit(_inbound.putIndex());
+
+ result=_engine.unwrap(in_buffer,bbuf);
+ if (LOG.isDebugEnabled())
+ LOG.debug("{} unwrap {} {} consumed={} produced={}",
+ _session,
+ result.getStatus(),
+ result.getHandshakeStatus(),
+ result.bytesConsumed(),
+ result.bytesProduced());
+
+ _inbound.skip(result.bytesConsumed());
+ _inbound.compact();
+ buffer.setPutIndex(buffer.putIndex()+result.bytesProduced());
+ }
+ catch(SSLException e)
+ {
+ LOG.warn(_endp+":"+e);
+ LOG.debug(e);
+ if (_endp.isOpen())
+ _endp.close();
+ throw e;
+ }
+ finally
+ {
+ in_buffer.position(0);
+ in_buffer.limit(in_buffer.capacity());
+ bbuf.position(0);
+ bbuf.limit(bbuf.capacity());
+ }
+ }
+ }
+
+ switch(result.getStatus())
+ {
+ case BUFFER_UNDERFLOW:
+ break;
+
+ case BUFFER_OVERFLOW:
+ LOG.debug("{} unwrap {} {}->{}",_session,result.getStatus(),_inbound.toDetailString(),buffer.toDetailString());
+ break;
+
+ case OK:
+ if (result.getHandshakeStatus()==HandshakeStatus.FINISHED)
+ _handshook=true;
+ break;
+
+ case CLOSED:
+ LOG.debug("unwrap CLOSE {} {}",this,result);
+ if (result.getHandshakeStatus()==HandshakeStatus.FINISHED)
+ _endp.close();
+ break;
+
+ default:
+ LOG.warn("{} wrap default {}",_session,result);
+ throw new IOException(result.toString());
+ }
+
+ //if (LOG.isDebugEnabled() && result.bytesProduced()>0)
+ // LOG.debug("{} unwrapped '{}'",_session,buffer);
+
+ return result.bytesConsumed()>0 || result.bytesProduced()>0;
+ }
+
+
+ /* ------------------------------------------------------------ */
+ private ByteBuffer extractByteBuffer(Buffer buffer)
+ {
+ if (buffer.buffer() instanceof NIOBuffer)
+ return ((NIOBuffer)buffer.buffer()).getByteBuffer();
+ return ByteBuffer.wrap(buffer.array());
+ }
+
+ /* ------------------------------------------------------------ */
+ public AsyncEndPoint getSslEndPoint()
+ {
+ return _sslEndPoint;
+ }
+
+ public String toString()
+ {
+ return String.format("%s | %s", super.toString(), _sslEndPoint);
+ }
+
+ /* ------------------------------------------------------------ */
+ /* ------------------------------------------------------------ */
+ public class SslEndPoint implements AsyncEndPoint
+ {
+ public SSLEngine getSslEngine()
+ {
+ return _engine;
+ }
+
+ public void shutdownOutput() throws IOException
+ {
+ synchronized (SslConnection.this)
+ {
+ LOG.debug("{} ssl endp.oshut {}",_session,this);
+ _engine.closeOutbound();
+ _oshut=true;
+ }
+ flush();
+ }
+
+ public boolean isOutputShutdown()
+ {
+ synchronized (SslConnection.this)
+ {
+ return _oshut||!isOpen()||_engine.isOutboundDone();
+ }
+ }
+
+ public void shutdownInput() throws IOException
+ {
+ LOG.debug("{} ssl endp.ishut!",_session);
+ // We do not do a closeInput here, as SSL does not support half close.
+ // isInputShutdown works it out itself from buffer state and underlying endpoint state.
+ }
+
+ public boolean isInputShutdown()
+ {
+ synchronized (SslConnection.this)
+ {
+ return _endp.isInputShutdown() &&
+ !(_unwrapBuf!=null&&_unwrapBuf.hasContent()) &&
+ !(_inbound!=null&&_inbound.hasContent());
+ }
+ }
+
+ public void close() throws IOException
+ {
+ LOG.debug("{} ssl endp.close",_session);
+ _endp.close();
+ }
+
+ public int fill(Buffer buffer) throws IOException
+ {
+ int size=buffer.length();
+ process(buffer,null);
+
+ int filled=buffer.length()-size;
+
+ if (filled==0 && isInputShutdown())
+ return -1;
+ return filled;
+ }
+
+ public int flush(Buffer buffer) throws IOException
+ {
+ int size = buffer.length();
+ process(null, buffer);
+ return size-buffer.length();
+ }
+
+ public int flush(Buffer header, Buffer buffer, Buffer trailer) throws IOException
+ {
+ if (header!=null && header.hasContent())
+ return flush(header);
+ if (buffer!=null && buffer.hasContent())
+ return flush(buffer);
+ if (trailer!=null && trailer.hasContent())
+ return flush(trailer);
+ return 0;
+ }
+
+ public boolean blockReadable(long millisecs) throws IOException
+ {
+ long now = System.currentTimeMillis();
+ long end=millisecs>0?(now+millisecs):Long.MAX_VALUE;
+
+ while (now
- * A SelectChannelEndPoint that uses an {@link SSLEngine} to handle an
- * SSL connection.
- *
- * There is a named logger "org.eclipse.jetty.http.ssl"
- *
- */
-public class SslSelectChannelEndPoint extends SelectChannelEndPoint
-{
- public static final Logger LOG=Log.getLogger("org.eclipse.jetty.io.nio").getLogger("ssl");
-
- private static final Buffer __EMPTY_BUFFER=new DirectNIOBuffer(0);
- private static final ByteBuffer __ZERO_BUFFER=ByteBuffer.allocate(0);
-
- private final Buffers _buffers;
-
- private final SSLEngine _engine;
- private final SSLSession _session;
- private volatile NIOBuffer _inNIOBuffer;
- private volatile NIOBuffer _outNIOBuffer;
-
- private boolean _closing=false;
- private SSLEngineResult _result;
-
- private volatile boolean _handshook=false;
- private boolean _allowRenegotiate=true;
-
- private volatile boolean _debug = LOG.isDebugEnabled(); // snapshot debug status for optimizer
-
- /* ------------------------------------------------------------ */
- public SslSelectChannelEndPoint(Buffers buffers,SocketChannel channel, SelectorManager.SelectSet selectSet, SelectionKey key, SSLEngine engine, int maxIdleTime)
- throws IOException
- {
- super(channel,selectSet,key, maxIdleTime);
- _buffers=buffers;
-
- // ssl
- _engine=engine;
- _session=engine.getSession();
-
- if (_debug) LOG.debug(_session+" channel="+channel);
- }
-
- /* ------------------------------------------------------------ */
- public SslSelectChannelEndPoint(Buffers buffers,SocketChannel channel, SelectorManager.SelectSet selectSet, SelectionKey key, SSLEngine engine)
- throws IOException
- {
- super(channel,selectSet,key);
- _buffers=buffers;
-
- // ssl
- _engine=engine;
- _session=engine.getSession();
-
- if (_debug) LOG.debug(_session+" channel="+channel);
- }
-
-
- /* ------------------------------------------------------------ */
- private void needOutBuffer()
- {
- synchronized (this)
- {
- if (_outNIOBuffer==null)
- _outNIOBuffer=(NIOBuffer)_buffers.getBuffer(_session.getPacketBufferSize());
- }
- }
-
- /* ------------------------------------------------------------ */
- private void freeOutBuffer()
- {
- synchronized (this)
- {
- if (_outNIOBuffer!=null && _outNIOBuffer.length()==0)
- {
- _buffers.returnBuffer(_outNIOBuffer);
- _outNIOBuffer=null;
- }
- }
- }
-
- /* ------------------------------------------------------------ */
- private void needInBuffer()
- {
- synchronized (this)
- {
- if(_inNIOBuffer==null)
- _inNIOBuffer=(NIOBuffer)_buffers.getBuffer(_session.getPacketBufferSize());
- }
- }
-
- /* ------------------------------------------------------------ */
- private void freeInBuffer()
- {
- synchronized (this)
- {
- if (_inNIOBuffer!=null && _inNIOBuffer.length()==0)
- {
- _buffers.returnBuffer(_inNIOBuffer);
- _inNIOBuffer=null;
- }
- }
- }
-
- /* ------------------------------------------------------------ */
- /**
- * @return True if the endpoint has produced/consumed bytes itself (non application data).
- */
- public boolean isProgressing()
- {
- SSLEngineResult result = _result;
- _result=null;
- return result!=null && (result.bytesConsumed()>0 || result.bytesProduced()>0);
- }
-
- /* ------------------------------------------------------------ */
- /**
- * @return True if SSL re-negotiation is allowed (default false)
- */
- public boolean isAllowRenegotiate()
- {
- return _allowRenegotiate;
- }
-
- /* ------------------------------------------------------------ */
- /**
- * Set if SSL re-negotiation is allowed. CVE-2009-3555 discovered
- * a vulnerability in SSL/TLS with re-negotiation. If your JVM
- * does not have CVE-2009-3555 fixed, then re-negotiation should
- * not be allowed.
- * @param allowRenegotiate true if re-negotiation is allowed (default false)
- */
- public void setAllowRenegotiate(boolean allowRenegotiate)
- {
- _allowRenegotiate = allowRenegotiate;
- }
-
-
- /* ------------------------------------------------------------ */
- @Override
- public boolean isOutputShutdown()
- {
- return _engine!=null && _engine.isOutboundDone();
- }
-
- /* ------------------------------------------------------------ */
- @Override
- public boolean isInputShutdown()
- {
- return _engine!=null && _engine.isInboundDone();
- }
-
- /* ------------------------------------------------------------ */
- @Override
- public void shutdownOutput() throws IOException
- {
- LOG.debug("{} shutdownOutput",_session);
- // All SSL closes should be graceful, as it is more secure.
- // So normal SSL close can be used here.
- close();
- }
-
- /* ------------------------------------------------------------ */
- private int process(ByteBuffer inBBuf, Buffer outBuf) throws IOException
- {
- if (_debug)
- LOG.debug("{} process closing={} in={} out={}",_session,_closing,inBBuf,outBuf);
-
- // If there is no place to put incoming application data,
- if (inBBuf==null)
- {
- // use ZERO buffer
- inBBuf=__ZERO_BUFFER;
- }
-
- int received=0;
- int sent=0;
-
-
- HandshakeStatus initialStatus = _engine.getHandshakeStatus();
- boolean progress=true;
-
- while (progress)
- {
- progress=false;
-
- // flush output data
- int len=_outNIOBuffer==null?0:_outNIOBuffer.length();
-
- // we must flush it, as the other end might be
- // waiting for that outgoing data before sending
- // more incoming data
- flush();
-
- // If we have written some bytes, then progress has been made.
- progress|=(_outNIOBuffer==null?0:_outNIOBuffer.length())0||_result.bytesProduced()>0||_result.bytesConsumed()>0;
-
- if (c>0)
- sent+=c;
- else if (c<0 && sent==0)
- sent=-1;
- }
-
- // Try unwrapping some application data
- if (inBBuf.remaining()>0 && _inNIOBuffer!=null && _inNIOBuffer.hasContent())
- {
- int space=inBBuf.remaining();
- progress|=unwrap(inBBuf);
- received+=space-inBBuf.remaining();
- }
- break;
-
-
- case NEED_TASK:
- {
- // A task needs to be run, so run it!
- Runnable task;
- while ((task=_engine.getDelegatedTask())!=null)
- {
- progress=true;
- task.run();
- }
-
- // Detect SUN JVM Bug!!!
- if(initialStatus==HandshakeStatus.NOT_HANDSHAKING &&
- _engine.getHandshakeStatus()==HandshakeStatus.NEED_UNWRAP && sent==0)
- {
- // This should be NEED_WRAP
- // The fix simply detects the signature of the bug and then close the connection (fail-fast) so that ff3 will delegate to using SSL instead of TLS.
- // This is a jvm bug on java1.6 where the SSLEngine expects more data from the initial handshake when the client(ff3-tls) already had given it.
- // See http://jira.codehaus.org/browse/JETTY-567 for more details
- if (_debug) LOG.warn("{} JETTY-567",_session);
- return -1;
- }
- break;
- }
-
- case NEED_WRAP:
- {
- checkRenegotiate();
-
- // The SSL needs to send some handshake data to the other side
- int c=0;
- if (outBuf!=null && outBuf.hasContent())
- c=wrap(outBuf);
- else
- c=wrap(__EMPTY_BUFFER);
-
- progress=_result.bytesProduced()>0||_result.bytesConsumed()>0;
- if (c>0)
- sent+=c;
- else if (c<0 && sent==0)
- sent=-1;
- break;
- }
-
- case NEED_UNWRAP:
- {
- checkRenegotiate();
-
- // Need more data to be unwrapped so try another call to unwrap
- progress|=unwrap(inBBuf);
- if (_closing && inBBuf.hasRemaining())
- inBBuf.clear();
- break;
- }
- }
-
- if (_debug) LOG.debug("{} progress {}",_session,progress);
- }
-
- if (_debug) LOG.debug("{} received {} sent {}",_session,received,sent);
-
- freeInBuffer();
- return (received<0||sent<0)?-1:(received+sent);
- }
-
-
-
- /* ------------------------------------------------------------ */
- @Override
- public void close() throws IOException
- {
- // For safety we always force a close calling super
- try
- {
- if (!_closing)
- {
- _closing=true;
- LOG.debug("{} close",_session);
- _engine.closeOutbound();
- process(null,null);
- }
- }
- catch (IOException e)
- {
- // We could not write the SSL close message because the
- // socket was already closed, nothing more we can do.
- LOG.ignore(e);
- }
- finally
- {
- super.close();
- }
- }
-
- /* ------------------------------------------------------------ */
- /** Fill the buffer with unencrypted bytes.
- * Called by a Http Parser when more data is
- * needed to continue parsing a request or a response.
- */
- @Override
- public int fill(Buffer buffer) throws IOException
- {
- _debug=LOG.isDebugEnabled();
- LOG.debug("{} fill",_session);
- // This end point only works on NIO buffer type (director
- // or indirect), so extract the NIO buffer that is wrapped
- // by the passed jetty Buffer.
- ByteBuffer bbuf=((NIOBuffer)buffer).getByteBuffer();
-
-
- // remember the original size of the unencrypted buffer
- int size=buffer.length();
-
-
- synchronized (bbuf)
- {
- bbuf.position(buffer.putIndex());
- try
- {
- // Call the SSLEngine unwrap method to process data in
- // the inBuffer. If there is no data in the inBuffer, then
- // super.fill is called to read encrypted bytes.
- unwrap(bbuf);
- process(bbuf,null);
- }
- finally
- {
- // reset the Buffers
- buffer.setPutIndex(bbuf.position());
- bbuf.position(0);
- }
- }
- // return the number of unencrypted bytes filled.
- int filled=buffer.length()-size;
- if (filled==0 && (isInputShutdown() || !isOpen()))
- return -1;
-
- return filled;
- }
-
- /* ------------------------------------------------------------ */
- @Override
- public int flush(Buffer buffer) throws IOException
- {
- _debug=LOG.isDebugEnabled();
- LOG.debug("{} flush1",_session);
- return process(null,buffer);
- }
-
-
- /* ------------------------------------------------------------ */
- /*
- */
- @Override
- public int flush(Buffer header, Buffer buffer, Buffer trailer) throws IOException
- {
- _debug=LOG.isDebugEnabled();
- LOG.debug("{} flush3",_session);
-
- int len=0;
- int flushed=0;
- if (header!=null && header.hasContent())
- {
- len=header.length();
- flushed=flush(header);
- }
- if (flushed==len && buffer!=null && buffer.hasContent())
- {
- int f=flush(buffer);
- if (f>=0)
- flushed+=f;
- else if (flushed==0)
- flushed=-1;
- }
-
- return flushed;
- }
-
- /* ------------------------------------------------------------ */
- @Override
- public void flush() throws IOException
- {
- LOG.debug("{} flush",_session);
- if (!isOpen())
- throw new EofException();
-
- if (isBufferingOutput())
- {
- int flushed=super.flush(_outNIOBuffer);
- if (_debug)
- LOG.debug("{} flushed={} left={}",_session,flushed,_outNIOBuffer.length());
- }
- else if (_engine.isOutboundDone() && super.isOpen())
- {
- if (_debug)
- LOG.debug("{} flush shutdownOutput",_session);
- try
- {
- super.shutdownOutput();
- }
- catch(IOException e)
- {
- LOG.ignore(e);
- }
- }
-
- freeOutBuffer();
- }
-
- /* ------------------------------------------------------------ */
- private void checkRenegotiate() throws IOException
- {
- if (_handshook && !_allowRenegotiate && _channel!=null && _channel.isOpen())
- {
- LOG.warn("SSL renegotiate denied: {}",_channel);
- super.close();
- }
- }
-
- /* ------------------------------------------------------------ */
- /**
- * @return true if progress is made
- */
- private boolean unwrap(ByteBuffer buffer) throws IOException
- {
- needInBuffer();
- ByteBuffer in_buffer=_inNIOBuffer.getByteBuffer();
-
- _inNIOBuffer.compact();
-
- int total_filled=0;
- boolean remoteClosed = false;
-
- LOG.debug("{} unwrap space={} open={}",_session,_inNIOBuffer.space(),super.isOpen());
-
- // loop filling as much encrypted data as we can into the buffer
- while (_inNIOBuffer.space()>0 && super.isOpen())
- {
- int filled=super.fill(_inNIOBuffer);
- if (_debug) LOG.debug("{} filled {}",_session,filled);
- if (filled < 0)
- remoteClosed = true;
- // break the loop if no progress is made (we have read everything there is to read)
- if (filled<=0)
- break;
- total_filled+=filled;
- }
-
- // If we have no progress and no data
- if (total_filled==0 && _inNIOBuffer.length()==0)
- {
- // Do we need to close?
- if (isOpen() && remoteClosed)
- {
- try
- {
- _engine.closeInbound();
- }
- catch (SSLException x)
- {
- // It may happen, for example, in case of truncation
- // attacks, we close so that we do not spin forever
- super.close();
- }
- }
-
- if (!isOpen())
- throw new EofException();
-
- return false;
- }
-
- // We have some in data, so try to unwrap it.
- try
- {
- // inBuffer is the NIO buffer inside the _inNIOBuffer,
- // so update its position and limit from the inNIOBuffer.
- in_buffer.position(_inNIOBuffer.getIndex());
- in_buffer.limit(_inNIOBuffer.putIndex());
-
- // Do the unwrap
- _result=_engine.unwrap(in_buffer,buffer);
- if (!_handshook && _result.getHandshakeStatus()==SSLEngineResult.HandshakeStatus.FINISHED)
- _handshook=true;
- if (_debug) LOG.debug("{} unwrap {}",_session,_result);
-
- // skip the bytes consumed
- _inNIOBuffer.skip(_result.bytesConsumed());
- }
- catch(SSLException e)
- {
- LOG.debug(getRemoteAddr() + ":" + getRemotePort() + " ",e);
- super.close();
- throw e;
- }
- finally
- {
- // reset the buffer so it can be managed by the _inNIOBuffer again.
- in_buffer.position(0);
- in_buffer.limit(in_buffer.capacity());
- }
-
- // handle the unwrap results
- switch(_result.getStatus())
- {
- case BUFFER_OVERFLOW:
- LOG.debug("{} unwrap overflow",_session);
- return false;
-
- case BUFFER_UNDERFLOW:
- // Not enough data,
- // If we are closed, we will never get more, so EOF
- // else return and we will be tried again
- // later when more data arriving causes another dispatch.
- if (LOG.isDebugEnabled()) LOG.debug("{} unwrap {}",_session,_result);
- if(!isOpen())
- {
- _inNIOBuffer.clear();
- if (_outNIOBuffer!=null)
- _outNIOBuffer.clear();
- throw new EofException();
- }
- return (total_filled > 0);
-
- case CLOSED:
- _closing=true;
- // return true is some bytes somewhere were moved about.
- return total_filled>0 ||_result.bytesConsumed()>0 || _result.bytesProduced()>0;
-
- case OK:
- // return true is some bytes somewhere were moved about.
- return total_filled>0 ||_result.bytesConsumed()>0 || _result.bytesProduced()>0;
-
- default:
- LOG.warn("{} unwrap default: {}",_session,_result);
- throw new IOException(_result.toString());
- }
- }
-
- /* ------------------------------------------------------------ */
- private ByteBuffer extractOutputBuffer(Buffer buffer)
- {
- if (buffer.buffer() instanceof NIOBuffer)
- return ((NIOBuffer)buffer.buffer()).getByteBuffer();
-
- return ByteBuffer.wrap(buffer.array());
- }
-
- /* ------------------------------------------------------------ */
- private int wrap(final Buffer buffer) throws IOException
- {
- ByteBuffer bbuf=extractOutputBuffer(buffer);
- synchronized(bbuf)
- {
- int consumed=0;
- needOutBuffer();
- _outNIOBuffer.compact();
- ByteBuffer out_buffer=_outNIOBuffer.getByteBuffer();
- synchronized(out_buffer)
- {
- try
- {
- bbuf.position(buffer.getIndex());
- bbuf.limit(buffer.putIndex());
- out_buffer.position(_outNIOBuffer.putIndex());
- out_buffer.limit(out_buffer.capacity());
- _result=_engine.wrap(bbuf,out_buffer);
- if (_debug) LOG.debug("{} wrap {}",_session,_result);
- if (!_handshook && _result.getHandshakeStatus()==SSLEngineResult.HandshakeStatus.FINISHED)
- _handshook=true;
- _outNIOBuffer.setPutIndex(out_buffer.position());
- consumed=_result.bytesConsumed();
- }
- catch(SSLException e)
- {
- LOG.warn(getRemoteAddr()+":"+getRemotePort()+" ",e);
- if (getChannel().isOpen())
- getChannel().close();
- throw e;
- }
- finally
- {
- out_buffer.position(0);
- bbuf.position(0);
- bbuf.limit(bbuf.capacity());
-
- if (consumed>0)
- {
- int len=consumed0?_result.bytesConsumed():-1;
-
- default:
- LOG.warn("{} wrap default {}",_session,_result);
- throw new IOException(_result.toString());
- }
- }
-
- /* ------------------------------------------------------------ */
- @Override
- public boolean isBufferingInput()
- {
- final Buffer in = _inNIOBuffer;
- return in!=null && in.hasContent();
- }
-
- /* ------------------------------------------------------------ */
- @Override
- public boolean isBufferingOutput()
- {
- final NIOBuffer out = _outNIOBuffer;
- return out!=null && out.hasContent();
- }
-
- /* ------------------------------------------------------------ */
- @Override
- public boolean isBufferred()
- {
- return true;
- }
-
- /* ------------------------------------------------------------ */
- public SSLEngine getSSLEngine()
- {
- return _engine;
- }
-
- /* ------------------------------------------------------------ */
- @Override
- public String toString()
- {
- final NIOBuffer i=_inNIOBuffer;
- final NIOBuffer o=_outNIOBuffer;
- return "SSL"+super.toString()+","+(_engine==null?"-":_engine.getHandshakeStatus())+", in/out="+
- (i==null?0:i.length())+"/"+(o==null?0:o.length())+
- " bi/o="+isBufferingInput()+"/"+isBufferingOutput()+
- " "+_result;
- }
-}
diff --git a/jetty-io/src/test/java/org/eclipse/jetty/io/EndPointTest.java b/jetty-io/src/test/java/org/eclipse/jetty/io/EndPointTest.java
index 75922f50437..51be3cd751f 100644
--- a/jetty-io/src/test/java/org/eclipse/jetty/io/EndPointTest.java
+++ b/jetty-io/src/test/java/org/eclipse/jetty/io/EndPointTest.java
@@ -1,62 +1,157 @@
package org.eclipse.jetty.io;
-import java.net.ServerSocket;
-import java.net.Socket;
-import java.util.concurrent.Exchanger;
-import java.util.concurrent.TimeUnit;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
-import org.eclipse.jetty.io.bio.SocketEndPoint;
import org.eclipse.jetty.io.nio.IndirectNIOBuffer;
-import org.junit.Assert;
import org.junit.Test;
-public class EndPointTest
+public abstract class EndPointTest
{
+ public static class EndPointPair
+ {
+ public T client;
+ public T server;
+ }
+
+ protected abstract EndPointPair newConnection() throws Exception;
+
+
@Test
- public void testSocketEndPoints() throws Exception
+ public void testClientServerExchange() throws Exception
{
- final ServerSocket server = new ServerSocket();
- server.bind(null);
+ EndPointPair c = newConnection();
+ Buffer buffer = new IndirectNIOBuffer(4096);
- final Exchanger accepted = new Exchanger();
- new Thread(){
- public void run()
- {
- try
- {
- accepted.exchange(server.accept());
- }
- catch(Exception e)
- {
- e.printStackTrace();
- }
- }
- }.start();
+ // Client sends a request
+ c.client.flush(new ByteArrayBuffer("request"));
- Socket s0 = new Socket(server.getInetAddress(),server.getLocalPort());
- Socket s1 = accepted.exchange(null,5,TimeUnit.SECONDS);
+ // Server receives the request
+ int len = c.server.fill(buffer);
+ assertEquals(7,len);
+ assertEquals("request",buffer.toString());
+
+ // Client and server are open
+ assertTrue(c.client.isOpen());
+ assertFalse(c.client.isInputShutdown());
+ assertFalse(c.client.isOutputShutdown());
+ assertTrue(c.server.isOpen());
+ assertFalse(c.server.isInputShutdown());
+ assertFalse(c.server.isOutputShutdown());
- SocketEndPoint in = new SocketEndPoint(s0);
- SocketEndPoint out = new SocketEndPoint(s1);
+ // Server sends response and closes output
+ c.server.flush(new ByteArrayBuffer("response"));
+ c.server.shutdownOutput();
+
+ // client server are open, server is oshut
+ assertTrue(c.client.isOpen());
+ assertFalse(c.client.isInputShutdown());
+ assertFalse(c.client.isOutputShutdown());
+ assertTrue(c.server.isOpen());
+ assertFalse(c.server.isInputShutdown());
+ assertTrue(c.server.isOutputShutdown());
+
+ // Client reads response
+ buffer.clear();
+ len = c.client.fill(buffer);
+ assertEquals(8,len);
+ assertEquals("response",buffer.toString());
+
+ // Client and server are open, server is oshut
+ assertTrue(c.client.isOpen());
+ assertFalse(c.client.isInputShutdown());
+ assertFalse(c.client.isOutputShutdown());
+ assertTrue(c.server.isOpen());
+ assertFalse(c.server.isInputShutdown());
+ assertTrue(c.server.isOutputShutdown());
+
+ // Client reads -1
+ buffer.clear();
+ len = c.client.fill(buffer);
+ assertEquals(-1,len);
+
+ // Client and server are open, server is oshut, client is ishut
+ assertTrue(c.client.isOpen());
+ assertTrue(c.client.isInputShutdown());
+ assertFalse(c.client.isOutputShutdown());
+ assertTrue(c.server.isOpen());
+ assertFalse(c.server.isInputShutdown());
+ assertTrue(c.server.isOutputShutdown());
+
+ // Client shutsdown output, which is a close because already ishut
+ c.client.shutdownOutput();
+
+ // Client is closed. Server is open and oshut
+ assertFalse(c.client.isOpen());
+ assertTrue(c.client.isInputShutdown());
+ assertTrue(c.client.isOutputShutdown());
+ assertTrue(c.server.isOpen());
+ assertFalse(c.server.isInputShutdown());
+ assertTrue(c.server.isOutputShutdown());
+
+ // Server reads close
+ buffer.clear();
+ len = c.server.fill(buffer);
+ assertEquals(-1,len);
+
+ // Client and Server are closed
+ assertFalse(c.client.isOpen());
+ assertTrue(c.client.isInputShutdown());
+ assertTrue(c.client.isOutputShutdown());
+ assertFalse(c.server.isOpen());
+ assertTrue(c.server.isInputShutdown());
+ assertTrue(c.server.isOutputShutdown());
- check(in,out);
}
-
- private void check(EndPoint in, EndPoint out) throws Exception
+
+
+ @Test
+ public void testClientClose() throws Exception
{
- String data="Now is the time for all good men to come to the aid of the party";
- Buffer send = new ByteArrayBuffer(data);
- Buffer receive = new IndirectNIOBuffer(4096);
+ EndPointPair c = newConnection();
+ Buffer buffer = new IndirectNIOBuffer(4096);
- int lo=out.flush(send);
- int li=in.fill(receive);
+ c.client.flush(new ByteArrayBuffer("request"));
+ int len = c.server.fill(buffer);
+ assertEquals(7,len);
+ assertEquals("request",buffer.toString());
+
+ assertTrue(c.client.isOpen());
+ assertFalse(c.client.isInputShutdown());
+ assertFalse(c.client.isOutputShutdown());
+ assertTrue(c.server.isOpen());
+ assertFalse(c.server.isInputShutdown());
+ assertFalse(c.server.isOutputShutdown());
- Assert.assertEquals(data.length(),lo);
- Assert.assertEquals(data.length(),li);
- Assert.assertEquals(data,receive.toString());
+ c.client.close();
+
+ assertFalse(c.client.isOpen());
+ assertTrue(c.client.isInputShutdown());
+ assertTrue(c.client.isOutputShutdown());
+ assertTrue(c.server.isOpen());
+ assertFalse(c.server.isInputShutdown());
+ assertFalse(c.server.isOutputShutdown());
- in.close();
- out.close();
- }
+ len = c.server.fill(buffer);
+ assertEquals(-1,len);
+
+ assertFalse(c.client.isOpen());
+ assertTrue(c.client.isInputShutdown());
+ assertTrue(c.client.isOutputShutdown());
+ assertTrue(c.server.isOpen());
+ assertTrue(c.server.isInputShutdown());
+ assertFalse(c.server.isOutputShutdown());
+
+ c.server.shutdownOutput();
+
+ assertFalse(c.client.isOpen());
+ assertTrue(c.client.isInputShutdown());
+ assertTrue(c.client.isOutputShutdown());
+ assertFalse(c.server.isOpen());
+ assertTrue(c.server.isInputShutdown());
+ assertTrue(c.server.isOutputShutdown());
+ }
+
}
diff --git a/jetty-io/src/test/java/org/eclipse/jetty/io/IOTest.java b/jetty-io/src/test/java/org/eclipse/jetty/io/IOTest.java
index 1ca3b75fdfc..8575ab7f229 100644
--- a/jetty-io/src/test/java/org/eclipse/jetty/io/IOTest.java
+++ b/jetty-io/src/test/java/org/eclipse/jetty/io/IOTest.java
@@ -14,15 +14,17 @@
package org.eclipse.jetty.io;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
-import java.io.Reader;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
import org.eclipse.jetty.util.IO;
import org.junit.Test;
@@ -50,14 +52,12 @@ public class IOTest
}
@Test
- public void testHalfCloses() throws Exception
+ public void testHalfClose() throws Exception
{
ServerSocket connector = new ServerSocket(0);
Socket client = new Socket("localhost",connector.getLocalPort());
- System.err.println(client);
Socket server = connector.accept();
- System.err.println(server);
// we can write both ways
client.getOutputStream().write(1);
@@ -76,7 +76,7 @@ public class IOTest
assertEquals(-1,server.getInputStream().read());
// but cannot write
- try { client.getOutputStream().write(1); assertTrue(false); } catch (SocketException e) {}
+ try { client.getOutputStream().write(1); fail("exception expected"); } catch (SocketException e) {}
// but can still write in opposite direction.
server.getOutputStream().write(1);
@@ -87,7 +87,7 @@ public class IOTest
server.shutdownInput();
// now we EOF instead of reading -1
- try { server.getInputStream().read(); assertTrue(false); } catch (SocketException e) {}
+ try { server.getInputStream().read(); fail("exception expected"); } catch (SocketException e) {}
// but can still write in opposite direction.
@@ -98,7 +98,7 @@ public class IOTest
client.shutdownInput();
// now we EOF instead of reading -1
- try { client.getInputStream().read(); assertTrue(false); } catch (SocketException e) {}
+ try { client.getInputStream().read(); fail("exception expected"); } catch (SocketException e) {}
// But we can still write at the server (data which will never be read)
server.getOutputStream().write(1);
@@ -110,7 +110,7 @@ public class IOTest
server.shutdownOutput();
// and now we can't write
- try { server.getOutputStream().write(1); assertTrue(false); } catch (SocketException e) {}
+ try { server.getOutputStream().write(1); fail("exception expected"); } catch (SocketException e) {}
// but the sockets are still open
assertFalse(client.isClosed());
@@ -128,8 +128,114 @@ public class IOTest
// which has to be closed explictly
server.close();
assertTrue(server.isClosed());
-
-
-
+
}
+
+
+ @Test
+ public void testHalfCloseClientServer() throws Exception
+ {
+ ServerSocketChannel connector = ServerSocketChannel.open();
+ connector.socket().bind(null);
+
+ Socket client = SocketChannel.open(connector.socket().getLocalSocketAddress()).socket();
+ client.setSoTimeout(1000);
+ client.setSoLinger(false,-1);
+ Socket server = connector.accept().socket();
+ server.setSoTimeout(1000);
+ server.setSoLinger(false,-1);
+
+ // Write from client to server
+ client.getOutputStream().write(1);
+
+ // Server reads
+ assertEquals(1,server.getInputStream().read());
+
+ // Write from server to client with oshut
+ server.getOutputStream().write(1);
+ System.err.println("OSHUT "+server);
+ server.shutdownOutput();
+
+ // Client reads response
+ assertEquals(1,client.getInputStream().read());
+
+ try
+ {
+ // Client reads -1 and does ishut
+ assertEquals(-1,client.getInputStream().read());
+ assertFalse(client.isInputShutdown());
+ System.err.println("ISHUT "+client);
+ client.shutdownInput();
+
+ // Client ???
+ System.err.println("OSHUT "+client);
+ client.shutdownOutput();
+ System.err.println("CLOSE "+client);
+ client.close();
+
+ // Server reads -1, does ishut and then close
+ assertEquals(-1,server.getInputStream().read());
+ assertFalse(server.isInputShutdown());
+ System.err.println("ISHUT "+server);
+
+ try
+ {
+ server.shutdownInput();
+ }
+ catch(SocketException e)
+ {
+ System.err.println(e);
+ }
+ System.err.println("CLOSE "+server);
+ server.close();
+
+ }
+ catch(Exception e)
+ {
+ // Dang OSX!
+ System.err.println(e);
+ }
+ }
+
+ @Test
+ public void testReset() throws Exception
+ {
+ ServerSocket connector;
+ Socket client;
+ Socket server;
+
+ connector = new ServerSocket(9123);
+ client = new Socket("127.0.0.1",connector.getLocalPort());
+ server = connector.accept();
+ client.setTcpNoDelay(true);
+ client.setSoLinger(true,0);
+ server.setTcpNoDelay(true);
+ server.setSoLinger(true,0);
+
+ client.getOutputStream().write(1);
+ assertEquals(1,server.getInputStream().read());
+ server.getOutputStream().write(1);
+ assertEquals(1,client.getInputStream().read());
+
+ // Server generator shutdowns output after non persistent sending response.
+ server.shutdownOutput();
+
+ // client endpoint reads EOF and shutdown input as result
+ assertEquals(-1,client.getInputStream().read());
+ client.shutdownInput();
+
+ // client connection see's EOF and shutsdown output as no more requests to be sent.
+ client.shutdownOutput();
+
+ // Since input already shutdown, client also closes socket.
+ client.close();
+
+ // Server reads the EOF from client oshut and shut's down it's input
+ assertEquals(-1,server.getInputStream().read());
+ server.shutdownInput();
+
+ // Since output was already shutdown, server closes
+ server.close();
+ }
+
}
diff --git a/jetty-io/src/test/java/org/eclipse/jetty/io/bio/SocketEndPointTest.java b/jetty-io/src/test/java/org/eclipse/jetty/io/bio/SocketEndPointTest.java
new file mode 100644
index 00000000000..277c6c7ccba
--- /dev/null
+++ b/jetty-io/src/test/java/org/eclipse/jetty/io/bio/SocketEndPointTest.java
@@ -0,0 +1,38 @@
+package org.eclipse.jetty.io.bio;
+
+import java.net.ServerSocket;
+import java.net.Socket;
+
+import org.eclipse.jetty.io.EndPointTest;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+public class SocketEndPointTest extends EndPointTest
+{
+ static ServerSocket connector;
+
+ @BeforeClass
+ public static void open() throws Exception
+ {
+ connector = new ServerSocket();
+ connector.bind(null);
+ }
+
+ @AfterClass
+ public static void close() throws Exception
+ {
+ connector.close();
+ connector=null;
+ }
+
+ @Override
+ protected EndPointPair newConnection() throws Exception
+ {
+ EndPointPair c = new EndPointPair();
+ c.client=new SocketEndPoint(new Socket(connector.getInetAddress(),connector.getLocalPort()));
+ c.server=new SocketEndPoint(connector.accept());
+ return c;
+ }
+
+
+}
diff --git a/jetty-io/src/test/java/org/eclipse/jetty/io/nio/ChannelEndPointTest.java b/jetty-io/src/test/java/org/eclipse/jetty/io/nio/ChannelEndPointTest.java
new file mode 100644
index 00000000000..1d8d5b2c7eb
--- /dev/null
+++ b/jetty-io/src/test/java/org/eclipse/jetty/io/nio/ChannelEndPointTest.java
@@ -0,0 +1,43 @@
+package org.eclipse.jetty.io.nio;
+
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+
+import org.eclipse.jetty.io.EndPointTest;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+public class ChannelEndPointTest extends EndPointTest
+{
+ static ServerSocketChannel connector;
+
+ @BeforeClass
+ public static void open() throws Exception
+ {
+ connector = ServerSocketChannel.open();
+ connector.socket().bind(null);
+ }
+
+ @AfterClass
+ public static void close() throws Exception
+ {
+ connector.close();
+ connector=null;
+ }
+
+ @Override
+ protected EndPointPair newConnection() throws Exception
+ {
+ EndPointPair c = new EndPointPair();
+
+ c.client=new ChannelEndPoint(SocketChannel.open(connector.socket().getLocalSocketAddress()));
+ c.server=new ChannelEndPoint(connector.accept());
+ return c;
+ }
+
+ @Override
+ public void testClientServerExchange() throws Exception
+ {
+ super.testClientServerExchange();
+ }
+}
diff --git a/jetty-io/src/test/java/org/eclipse/jetty/io/nio/NIOTest.java b/jetty-io/src/test/java/org/eclipse/jetty/io/nio/NIOTest.java
new file mode 100644
index 00000000000..9366d8ce843
--- /dev/null
+++ b/jetty-io/src/test/java/org/eclipse/jetty/io/nio/NIOTest.java
@@ -0,0 +1,131 @@
+// ========================================================================
+// Copyright (c) 2004-2009 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.io.nio;
+
+import static junit.framework.Assert.assertEquals;
+import static junit.framework.Assert.assertFalse;
+import static junit.framework.Assert.assertTrue;
+
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.ByteBuffer;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.SocketChannel;
+
+import org.junit.Test;
+
+/**
+ *
+ */
+public class NIOTest
+{
+ @Test
+ public void testSelector() throws Exception
+ {
+ ServerSocket acceptor = new ServerSocket(0);
+
+ Selector selector = Selector.open();
+
+ // Create client server socket pair
+ SocketChannel client = SocketChannel.open(acceptor.getLocalSocketAddress());
+ Socket server = acceptor.accept();
+ server.setTcpNoDelay(true);
+
+ // Make the client non blocking and register it with selector for reads
+ client.configureBlocking(false);
+ SelectionKey key = client.register(selector,SelectionKey.OP_READ);
+
+ // assert it is not selected
+ assertTrue(key.isValid());
+ assertFalse(key.isReadable());
+ assertEquals(0,key.readyOps());
+
+ // try selecting and assert nothing selected
+ int selected = selector.selectNow();
+ assertEquals(0,selected);
+ assertEquals(0,selector.selectedKeys().size());
+ assertTrue(key.isValid());
+ assertFalse(key.isReadable());
+ assertEquals(0,key.readyOps());
+
+ // Write a byte from server to client
+ server.getOutputStream().write(42);
+ server.getOutputStream().flush();
+
+ // select again and assert selection found for read
+ selected = selector.select(1000);
+ assertEquals(1,selected);
+ assertEquals(1,selector.selectedKeys().size());
+ assertTrue(key.isValid());
+ assertTrue(key.isReadable());
+ assertEquals(1,key.readyOps());
+
+ // select again and see that it is not reselect, but stays selected
+ selected = selector.select(100);
+ assertEquals(0,selected);
+ assertEquals(1,selector.selectedKeys().size());
+ assertTrue(key.isValid());
+ assertTrue(key.isReadable());
+ assertEquals(1,key.readyOps());
+
+ // read the byte
+ ByteBuffer buf = ByteBuffer.allocate(1024);
+ int len=client.read(buf);
+ assertEquals(1,len);
+ buf.flip();
+ assertEquals(42,buf.get());
+ buf.clear();
+
+ // But this does not change the key
+ assertTrue(key.isValid());
+ assertTrue(key.isReadable());
+ assertEquals(1,key.readyOps());
+
+ // Even if we select again ?
+ selected = selector.select(100);
+ assertEquals(0,selected);
+ assertEquals(1,selector.selectedKeys().size());
+ assertTrue(key.isValid());
+ assertTrue(key.isReadable());
+ assertEquals(1,key.readyOps());
+
+ // Unless we remove the key from the select set
+ // and then it is still flagged as isReadable()
+ selector.selectedKeys().clear();
+ assertEquals(0,selector.selectedKeys().size());
+ assertTrue(key.isValid());
+ assertTrue(key.isReadable());
+ assertEquals(1,key.readyOps());
+
+ // Now if we select again - it is still flagged as readable!!!
+ selected = selector.select(100);
+ assertEquals(0,selected);
+ assertEquals(0,selector.selectedKeys().size());
+ assertTrue(key.isValid());
+ assertTrue(key.isReadable());
+ assertEquals(1,key.readyOps());
+
+ // Only when it is selected for something else does that state change.
+ key.interestOps(SelectionKey.OP_READ|SelectionKey.OP_WRITE);
+ selected = selector.select(1000);
+ assertEquals(1,selected);
+ assertEquals(1,selector.selectedKeys().size());
+ assertTrue(key.isValid());
+ assertTrue(key.isWritable());
+ assertFalse(key.isReadable());
+ assertEquals(SelectionKey.OP_WRITE,key.readyOps());
+ }
+
+}
diff --git a/jetty-io/src/test/java/org/eclipse/jetty/io/nio/SelectChannelEndPointSslTest.java b/jetty-io/src/test/java/org/eclipse/jetty/io/nio/SelectChannelEndPointSslTest.java
new file mode 100644
index 00000000000..d6a6a46a06a
--- /dev/null
+++ b/jetty-io/src/test/java/org/eclipse/jetty/io/nio/SelectChannelEndPointSslTest.java
@@ -0,0 +1,187 @@
+package org.eclipse.jetty.io.nio;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.Socket;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLEngineResult;
+import javax.net.ssl.SSLEngineResult.HandshakeStatus;
+import javax.net.ssl.SSLSocket;
+
+import org.eclipse.jetty.io.EndPoint;
+import org.eclipse.jetty.toolchain.test.MavenTestingUtils;
+import org.eclipse.jetty.util.ssl.SslContextFactory;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+
+public class SelectChannelEndPointSslTest extends SelectChannelEndPointTest
+{
+ static SslContextFactory __sslCtxFactory=new SslContextFactory();
+
+ @BeforeClass
+ public static void initSslEngine() throws Exception
+ {
+ File keystore = MavenTestingUtils.getTestResourceFile("keystore");
+ __sslCtxFactory.setKeyStorePath(keystore.getAbsolutePath());
+ __sslCtxFactory.setKeyStorePassword("storepwd");
+ __sslCtxFactory.setKeyManagerPassword("keypwd");
+ __sslCtxFactory.start();
+ }
+
+ @Override
+ protected Socket newClient() throws IOException
+ {
+ SSLSocket socket = __sslCtxFactory.newSslSocket();
+ socket.connect(_connector.socket().getLocalSocketAddress());
+ return socket;
+ }
+
+ @Override
+ protected AsyncConnection newConnection(SocketChannel channel, EndPoint endpoint)
+ {
+ SSLEngine engine = __sslCtxFactory.newSslEngine();
+ engine.setUseClientMode(false);
+ SslConnection connection = new SslConnection(engine,endpoint);
+
+ AsyncConnection delegate = super.newConnection(channel,connection.getSslEndPoint());
+ connection.getSslEndPoint().setConnection(delegate);
+ return connection;
+ }
+
+ @Test
+ @Override
+ public void testEcho() throws Exception
+ {
+ super.testEcho();
+ }
+
+
+ @Test
+ @Override
+ public void testShutdown() throws Exception
+ {
+ // SSL does not do half closes
+ }
+
+ @Test
+ public void testTcpClose() throws Exception
+ {
+
+ // This test replaces SSLSocket() with a very manual SSL client
+ // so we can close TCP underneath SSL.
+
+ SocketChannel client = SocketChannel.open(_connector.socket().getLocalSocketAddress());
+ client.socket().setSoTimeout(500);
+
+ SocketChannel server = _connector.accept();
+ server.configureBlocking(false);
+ _manager.register(server);
+
+ SSLEngine engine = __sslCtxFactory.newSslEngine();
+ engine.setUseClientMode(true);
+ engine.beginHandshake();
+
+ ByteBuffer appOut = ByteBuffer.allocate(engine.getSession().getApplicationBufferSize());
+ ByteBuffer sslOut = ByteBuffer.allocate(engine.getSession().getPacketBufferSize()*2);
+ ByteBuffer appIn = ByteBuffer.allocate(engine.getSession().getApplicationBufferSize());
+ ByteBuffer sslIn = ByteBuffer.allocate(engine.getSession().getPacketBufferSize()*2);
+
+ boolean debug=SslConnection.LOG.isDebugEnabled();
+
+ if (debug) System.err.println(engine.getHandshakeStatus());
+ int loop=20;
+ while (engine.getHandshakeStatus()!=HandshakeStatus.NOT_HANDSHAKING)
+ {
+ if (--loop==0)
+ throw new IllegalStateException();
+
+ if (engine.getHandshakeStatus()==HandshakeStatus.NEED_WRAP)
+ {
+ if (debug) System.err.printf("sslOut %d-%d-%d%n",sslOut.position(),sslOut.limit(),sslOut.capacity());
+ if (debug) System.err.printf("appOut %d-%d-%d%n",appOut.position(),appOut.limit(),appOut.capacity());
+ SSLEngineResult result =engine.wrap(appOut,sslOut);
+ if (debug) System.err.println(result);
+ sslOut.flip();
+ int flushed=client.write(sslOut);
+ if (debug) System.err.println("out="+flushed);
+ sslOut.clear();
+ }
+
+ if (engine.getHandshakeStatus()==HandshakeStatus.NEED_UNWRAP)
+ {
+ if (debug) System.err.printf("sslIn %d-%d-%d%n",sslIn.position(),sslIn.limit(),sslIn.capacity());
+ if (sslIn.position()==0)
+ {
+ int filled=client.read(sslIn);
+ if (debug) System.err.println("in="+filled);
+ }
+ sslIn.flip();
+ if (debug) System.err.printf("sslIn %d-%d-%d%n",sslIn.position(),sslIn.limit(),sslIn.capacity());
+ SSLEngineResult result =engine.unwrap(sslIn,appIn);
+ if (debug) System.err.println(result);
+ if (debug) System.err.printf("sslIn %d-%d-%d%n",sslIn.position(),sslIn.limit(),sslIn.capacity());
+ if (sslIn.hasRemaining())
+ sslIn.compact();
+ else
+ sslIn.clear();
+ if (debug) System.err.printf("sslIn %d-%d-%d%n",sslIn.position(),sslIn.limit(),sslIn.capacity());
+ }
+
+ if (engine.getHandshakeStatus()==HandshakeStatus.NEED_TASK)
+ {
+ Runnable task;
+ while ((task=engine.getDelegatedTask())!=null)
+ task.run();
+ if (debug) System.err.println(engine.getHandshakeStatus());
+ }
+ }
+
+ if (debug) System.err.println("\nSay Hello");
+
+ // write a message
+ appOut.put("HelloWorld".getBytes("UTF-8"));
+ appOut.flip();
+ SSLEngineResult result =engine.wrap(appOut,sslOut);
+ if (debug) System.err.println(result);
+ sslOut.flip();
+ int flushed=client.write(sslOut);
+ if (debug) System.err.println("out="+flushed);
+ sslOut.clear();
+ appOut.clear();
+
+ // read the response
+ int filled=client.read(sslIn);
+ if (debug) System.err.println("in="+filled);
+ sslIn.flip();
+ result =engine.unwrap(sslIn,appIn);
+ if (debug) System.err.println(result);
+ if (sslIn.hasRemaining())
+ sslIn.compact();
+ else
+ sslIn.clear();
+
+ appIn.flip();
+ String reply= new String(appIn.array(),appIn.arrayOffset(),appIn.remaining());
+ appIn.clear();
+
+ Assert.assertEquals("HelloWorld",reply);
+
+ SelectorManager.LOG.info("javax.net.ssl.SSLException: Inbound closed... is expected soon");
+ if (debug) System.err.println("\nSudden Death");
+ client.socket().shutdownOutput();
+
+ filled=client.read(sslIn);
+ Assert.assertEquals(-1,filled);
+ }
+
+ @Test
+ public void testStress() throws Exception
+ {
+ super.testStress();
+ }
+}
diff --git a/jetty-io/src/test/java/org/eclipse/jetty/io/nio/SelectChannelEndPointTest.java b/jetty-io/src/test/java/org/eclipse/jetty/io/nio/SelectChannelEndPointTest.java
new file mode 100644
index 00000000000..768f872258e
--- /dev/null
+++ b/jetty-io/src/test/java/org/eclipse/jetty/io/nio/SelectChannelEndPointTest.java
@@ -0,0 +1,434 @@
+package org.eclipse.jetty.io.nio;
+
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.io.BufferedInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.net.SocketTimeoutException;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.eclipse.jetty.io.AbstractConnection;
+import org.eclipse.jetty.io.AsyncEndPoint;
+import org.eclipse.jetty.io.ConnectedEndPoint;
+import org.eclipse.jetty.io.EndPoint;
+import org.eclipse.jetty.util.thread.QueuedThreadPool;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class SelectChannelEndPointTest
+{
+ protected SelectChannelEndPoint _lastEndp;
+ protected ServerSocketChannel _connector;
+ protected QueuedThreadPool _threadPool = new QueuedThreadPool();
+ protected SelectorManager _manager = new SelectorManager()
+ {
+ @Override
+ public boolean dispatch(Runnable task)
+ {
+ return _threadPool.dispatch(task);
+ }
+
+ @Override
+ protected void endPointClosed(SelectChannelEndPoint endpoint)
+ {
+ }
+
+ @Override
+ protected void endPointOpened(SelectChannelEndPoint endpoint)
+ {
+ }
+
+ @Override
+ protected void endPointUpgraded(ConnectedEndPoint endpoint, org.eclipse.jetty.io.Connection oldConnection)
+ {
+ }
+
+ @Override
+ public AsyncConnection newConnection(SocketChannel channel, AsyncEndPoint endpoint, Object attachment)
+ {
+ return SelectChannelEndPointTest.this.newConnection(channel,endpoint);
+ }
+
+ @Override
+ protected SelectChannelEndPoint newEndPoint(SocketChannel channel, SelectSet selectSet, SelectionKey key) throws IOException
+ {
+ SelectChannelEndPoint endp = new SelectChannelEndPoint(channel,selectSet,key,2000);
+ endp.setConnection(selectSet.getManager().newConnection(channel,endp, key.attachment()));
+ _lastEndp=endp;
+ return endp;
+ }
+ };
+
+ // Must be volatile or the test may fail spuriously
+ private volatile int _blockAt=0;
+
+ @Before
+ public void startManager() throws Exception
+ {
+ _connector = ServerSocketChannel.open();
+ _connector.socket().bind(null);
+ _threadPool.start();
+ _manager.start();
+ }
+
+ @After
+ public void stopManager() throws Exception
+ {
+ _manager.stop();
+ _threadPool.stop();
+ _connector.close();
+ }
+
+ protected Socket newClient() throws IOException
+ {
+ return new Socket(_connector.socket().getInetAddress(),_connector.socket().getLocalPort());
+ }
+
+ protected AsyncConnection newConnection(SocketChannel channel, EndPoint endpoint)
+ {
+ return new TestConnection(endpoint);
+ }
+
+ public class TestConnection extends AbstractConnection implements AsyncConnection
+ {
+ NIOBuffer _in = new IndirectNIOBuffer(32*1024);
+ NIOBuffer _out = new IndirectNIOBuffer(32*1024);
+
+ public TestConnection(EndPoint endp)
+ {
+ super(endp);
+ }
+
+ public org.eclipse.jetty.io.Connection handle() throws IOException
+ {
+ boolean progress=true;
+ while(progress)
+ {
+ progress=false;
+ _in.compact();
+ if (_in.space()>0 && _endp.fill(_in)>0)
+ progress=true;
+
+ while (_blockAt>0 && _in.length()>0 && _in.length()<_blockAt)
+ {
+ _endp.blockReadable(10000);
+ if (_in.space()>0 && _endp.fill(_in)>0)
+ progress=true;
+ }
+
+ if (_in.hasContent() && _in.skip(_out.put(_in))>0)
+ progress=true;
+
+ if (_out.hasContent() && _endp.flush(_out)>0)
+ progress=true;
+
+ _out.compact();
+
+ if (!_out.hasContent() && _endp.isInputShutdown())
+ _endp.shutdownOutput();
+ }
+ return this;
+ }
+
+ public boolean isIdle()
+ {
+ return false;
+ }
+
+ public boolean isSuspended()
+ {
+ return false;
+ }
+
+ public void onClose()
+ {
+ // System.err.println("onClose");
+ }
+
+ public void onInputShutdown() throws IOException
+ {
+ // System.err.println("onInputShutdown");
+ }
+
+ }
+
+ @Test
+ public void testEcho() throws Exception
+ {
+ Socket client = newClient();
+
+ client.setSoTimeout(500);
+
+ SocketChannel server = _connector.accept();
+ server.configureBlocking(false);
+
+ _manager.register(server);
+
+ // Write client to server
+ client.getOutputStream().write("HelloWorld".getBytes("UTF-8"));
+
+ // Verify echo server to client
+ for (char c : "HelloWorld".toCharArray())
+ {
+ int b = client.getInputStream().read();
+ assertTrue(b>0);
+ assertEquals(c,(char)b);
+ }
+
+ // wait for read timeout
+ long start=System.currentTimeMillis();
+ try
+ {
+ client.getInputStream().read();
+ Assert.fail();
+ }
+ catch(SocketTimeoutException e)
+ {
+ assertTrue(System.currentTimeMillis()-start>=400);
+ }
+
+ // write then shutdown
+ client.getOutputStream().write("Goodbye Cruel TLS".getBytes("UTF-8"));
+
+ // Verify echo server to client
+ for (char c : "Goodbye Cruel TLS".toCharArray())
+ {
+ int b = client.getInputStream().read();
+ assertTrue(b>0);
+ assertEquals(c,(char)b);
+ }
+ client.close();
+
+ int i=0;
+ while (server.isOpen())
+ {
+ assert(i++<10);
+ Thread.sleep(10);
+ }
+
+ }
+
+
+ @Test
+ public void testShutdown() throws Exception
+ {
+ Socket client = newClient();
+
+ client.setSoTimeout(500);
+
+ SocketChannel server = _connector.accept();
+ server.configureBlocking(false);
+
+ _manager.register(server);
+
+ // Write client to server
+ client.getOutputStream().write("HelloWorld".getBytes("UTF-8"));
+
+ // Verify echo server to client
+ for (char c : "HelloWorld".toCharArray())
+ {
+ int b = client.getInputStream().read();
+ assertTrue(b>0);
+ assertEquals(c,(char)b);
+ }
+
+ // wait for read timeout
+ long start=System.currentTimeMillis();
+ try
+ {
+ client.getInputStream().read();
+ Assert.fail();
+ }
+ catch(SocketTimeoutException e)
+ {
+ assertTrue(System.currentTimeMillis()-start>=400);
+ }
+
+ // write then shutdown
+ client.getOutputStream().write("Goodbye Cruel TLS".getBytes("UTF-8"));
+ client.shutdownOutput();
+
+
+ // Verify echo server to client
+ for (char c : "Goodbye Cruel TLS".toCharArray())
+ {
+ int b = client.getInputStream().read();
+ assertTrue(b>0);
+ assertEquals(c,(char)b);
+ }
+
+ // Read close
+ assertEquals(-1,client.getInputStream().read());
+
+ }
+
+
+
+ @Test
+ public void testBlockIn() throws Exception
+ {
+ Socket client = newClient();
+
+ SocketChannel server = _connector.accept();
+ server.configureBlocking(false);
+
+ _manager.register(server);
+
+ OutputStream clientOutputStream = client.getOutputStream();
+ InputStream clientInputStream = client.getInputStream();
+
+ int specifiedTimeout = 400;
+ client.setSoTimeout(specifiedTimeout);
+
+ // Write 8 and cause block for 10
+ _blockAt=10;
+ clientOutputStream.write("12345678".getBytes("UTF-8"));
+ clientOutputStream.flush();
+
+ Thread.sleep(2 * specifiedTimeout);
+
+ // No echo as blocking for 10
+ long start=System.currentTimeMillis();
+ try
+ {
+ int b = clientInputStream.read();
+ Assert.fail("Should have timed out waiting for a response, but read "+b);
+ }
+ catch(SocketTimeoutException e)
+ {
+ int elapsed = Long.valueOf(System.currentTimeMillis() - start).intValue();
+ System.err.println("blocked for " + elapsed+ "ms");
+ Assert.assertThat("Expected timeout", elapsed, greaterThanOrEqualTo(3*specifiedTimeout/4));
+ }
+
+ // write remaining characters
+ clientOutputStream.write("90ABCDEF".getBytes("UTF-8"));
+ clientOutputStream.flush();
+
+ // Verify echo server to client
+ for (char c : "1234567890ABCDEF".toCharArray())
+ {
+ int b = clientInputStream.read();
+ assertTrue(b>0);
+ assertEquals(c,(char)b);
+ }
+ }
+
+ @Test
+ public void testIdle() throws Exception
+ {
+ Socket client = newClient();
+
+ client.setSoTimeout(3000);
+
+ SocketChannel server = _connector.accept();
+ server.configureBlocking(false);
+
+ _manager.register(server);
+
+ // Write client to server
+ client.getOutputStream().write("HelloWorld".getBytes("UTF-8"));
+
+ // Verify echo server to client
+ for (char c : "HelloWorld".toCharArray())
+ {
+ int b = client.getInputStream().read();
+ assertTrue(b>0);
+ assertEquals(c,(char)b);
+ }
+
+ // Set Max idle
+ _lastEndp.setMaxIdleTime(500);
+
+ // read until idle shutdown received
+ long start=System.currentTimeMillis();
+ int b=client.getInputStream().read();
+ assertEquals(-1,b);
+ long idle=System.currentTimeMillis()-start;
+ assertTrue(idle>400);
+ assertTrue(idle<2000);
+
+ // But endpoint is still open.
+ assertTrue(_lastEndp.isOpen());
+
+
+ // Wait for another idle callback
+ Thread.sleep(2000);
+ // endpoint is closed.
+
+ assertFalse(_lastEndp.isOpen());
+
+ }
+
+
+
+ @Test
+ public void testStress() throws Exception
+ {
+ Socket client = newClient();
+ client.setSoTimeout(30000);
+
+ SocketChannel server = _connector.accept();
+ server.configureBlocking(false);
+
+ _manager.register(server);
+ int writes = 100000;
+
+ final byte[] bytes="HelloWorld".getBytes("UTF-8");
+ final CountDownLatch latch = new CountDownLatch(writes);
+ final InputStream in = new BufferedInputStream(client.getInputStream());
+ final long start = System.currentTimeMillis();
+ client.getOutputStream().write(bytes);
+ client.getOutputStream().flush();
+
+ new Thread()
+ {
+ public void run()
+ {
+ try
+ {
+ while (latch.getCount()>0)
+ {
+ // Verify echo server to client
+ for (byte b0 : bytes)
+ {
+ int b = in.read();
+ assertTrue(b>0);
+ assertEquals(0xff&b0,b);
+ }
+ latch.countDown();
+ }
+ }
+ catch(Throwable e)
+ {
+ System.err.println("latch="+latch.getCount());
+ System.err.println("time="+(System.currentTimeMillis()-start));
+ e.printStackTrace();
+ }
+ }
+ }.start();
+
+
+ // Write client to server
+ for (int i=1;i
org.eclipse.jetty
jetty-project
- 7.5.4-SNAPSHOT
+ 7.6.0-SNAPSHOT
4.0.0
jetty-jaspi
diff --git a/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/callback/CredentialValidationCallback.java b/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/callback/CredentialValidationCallback.java
index e22aaa32e60..bf2c82c8c33 100644
--- a/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/callback/CredentialValidationCallback.java
+++ b/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/callback/CredentialValidationCallback.java
@@ -17,7 +17,7 @@ package org.eclipse.jetty.security.jaspi.callback;
import javax.security.auth.Subject;
import javax.security.auth.callback.Callback;
-import org.eclipse.jetty.http.security.Credential;
+import org.eclipse.jetty.util.security.Credential;
/**
* CredentialValidationCallback
diff --git a/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/modules/BaseAuthModule.java b/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/modules/BaseAuthModule.java
index 7099661a4da..4eb1ad6e256 100644
--- a/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/modules/BaseAuthModule.java
+++ b/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/modules/BaseAuthModule.java
@@ -32,8 +32,8 @@ import javax.security.auth.message.module.ServerAuthModule;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
-import org.eclipse.jetty.http.security.Credential;
-import org.eclipse.jetty.http.security.Password;
+import org.eclipse.jetty.util.security.Credential;
+import org.eclipse.jetty.util.security.Password;
import org.eclipse.jetty.security.authentication.LoginCallbackImpl;
import org.eclipse.jetty.security.jaspi.JaspiMessageInfo;
import org.eclipse.jetty.security.jaspi.callback.CredentialValidationCallback;
diff --git a/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/modules/BasicAuthModule.java b/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/modules/BasicAuthModule.java
index d97c5df7152..50032d46b26 100644
--- a/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/modules/BasicAuthModule.java
+++ b/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/modules/BasicAuthModule.java
@@ -27,7 +27,7 @@ import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.eclipse.jetty.http.HttpHeaders;
-import org.eclipse.jetty.http.security.Constraint;
+import org.eclipse.jetty.util.security.Constraint;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
diff --git a/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/modules/ClientCertAuthModule.java b/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/modules/ClientCertAuthModule.java
index 94300672200..5e637c4dedf 100644
--- a/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/modules/ClientCertAuthModule.java
+++ b/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/modules/ClientCertAuthModule.java
@@ -25,8 +25,8 @@ import javax.security.auth.message.MessageInfo;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
-import org.eclipse.jetty.http.security.Constraint;
-import org.eclipse.jetty.http.security.Password;
+import org.eclipse.jetty.util.security.Constraint;
+import org.eclipse.jetty.util.security.Password;
import org.eclipse.jetty.util.B64Code;
/**
diff --git a/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/modules/DigestAuthModule.java b/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/modules/DigestAuthModule.java
index 5702ee4892b..847b01b1c63 100644
--- a/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/modules/DigestAuthModule.java
+++ b/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/modules/DigestAuthModule.java
@@ -28,8 +28,8 @@ import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.eclipse.jetty.http.HttpHeaders;
-import org.eclipse.jetty.http.security.Constraint;
-import org.eclipse.jetty.http.security.Credential;
+import org.eclipse.jetty.util.security.Constraint;
+import org.eclipse.jetty.util.security.Credential;
import org.eclipse.jetty.util.B64Code;
import org.eclipse.jetty.util.QuotedStringTokenizer;
import org.eclipse.jetty.util.StringUtil;
diff --git a/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/modules/FormAuthModule.java b/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/modules/FormAuthModule.java
index 72fc9b7f55f..be390555742 100644
--- a/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/modules/FormAuthModule.java
+++ b/jetty-jaspi/src/main/java/org/eclipse/jetty/security/jaspi/modules/FormAuthModule.java
@@ -33,8 +33,8 @@ import javax.servlet.http.HttpSession;
import javax.servlet.http.HttpSessionBindingEvent;
import javax.servlet.http.HttpSessionBindingListener;
-import org.eclipse.jetty.http.security.Constraint;
-import org.eclipse.jetty.http.security.Password;
+import org.eclipse.jetty.util.security.Constraint;
+import org.eclipse.jetty.util.security.Password;
import org.eclipse.jetty.security.CrossContextPsuedoSession;
import org.eclipse.jetty.security.authentication.DeferredAuthentication;
import org.eclipse.jetty.security.authentication.LoginCallbackImpl;
diff --git a/jetty-jmx/pom.xml b/jetty-jmx/pom.xml
index 3b3f0b177b1..4b9a0dc2a9d 100644
--- a/jetty-jmx/pom.xml
+++ b/jetty-jmx/pom.xml
@@ -2,7 +2,7 @@
org.eclipse.jetty
jetty-project
- 7.5.4-SNAPSHOT
+ 7.6.0-SNAPSHOT
4.0.0
jetty-jmx
diff --git a/jetty-jmx/src/main/config/etc/jetty-jmx.xml b/jetty-jmx/src/main/config/etc/jetty-jmx.xml
index bdce9d42981..b8c38a201d6 100644
--- a/jetty-jmx/src/main/config/etc/jetty-jmx.xml
+++ b/jetty-jmx/src/main/config/etc/jetty-jmx.xml
@@ -9,6 +9,17 @@
+
+
+
+
+
+
diff --git a/jetty-jmx/src/main/java/org/eclipse/jetty/jmx/MBeanContainer.java b/jetty-jmx/src/main/java/org/eclipse/jetty/jmx/MBeanContainer.java
index 5371c8b60dc..d784442d9f4 100644
--- a/jetty-jmx/src/main/java/org/eclipse/jetty/jmx/MBeanContainer.java
+++ b/jetty-jmx/src/main/java/org/eclipse/jetty/jmx/MBeanContainer.java
@@ -22,11 +22,11 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.WeakHashMap;
+
import javax.management.MBeanServer;
import javax.management.ObjectInstance;
import javax.management.ObjectName;
-import org.eclipse.jetty.util.MultiMap;
import org.eclipse.jetty.util.component.AbstractLifeCycle;
import org.eclipse.jetty.util.component.AggregateLifeCycle;
import org.eclipse.jetty.util.component.Container;
@@ -41,7 +41,7 @@ import org.eclipse.jetty.util.thread.ShutdownThread;
*/
public class MBeanContainer extends AbstractLifeCycle implements Container.Listener, Dumpable
{
- private final static Logger __log = Log.getLogger(MBeanContainer.class.getName());
+ private final static Logger LOG = Log.getLogger(MBeanContainer.class.getName());
private final MBeanServer _server;
private final WeakHashMap _beans = new WeakHashMap();
@@ -93,7 +93,7 @@ public class MBeanContainer extends AbstractLifeCycle implements Container.Liste
}
catch (Exception e)
{
- __log.ignore(e);
+ LOG.ignore(e);
}
}
@@ -134,7 +134,7 @@ public class MBeanContainer extends AbstractLifeCycle implements Container.Liste
*/
public synchronized void add(Relationship relationship)
{
- __log.debug("add {}",relationship);
+ LOG.debug("add {}",relationship);
ObjectName parent = _beans.get(relationship.getParent());
if (parent == null)
{
@@ -168,7 +168,7 @@ public class MBeanContainer extends AbstractLifeCycle implements Container.Liste
*/
public synchronized void remove(Relationship relationship)
{
- __log.debug("remove {}",relationship);
+ LOG.debug("remove {}",relationship);
ObjectName parent = _beans.get(relationship.getParent());
ObjectName child = _beans.get(relationship.getChild());
@@ -194,7 +194,7 @@ public class MBeanContainer extends AbstractLifeCycle implements Container.Liste
*/
public synchronized void removeBean(Object obj)
{
- __log.debug("removeBean {}",obj);
+ LOG.debug("removeBean {}",obj);
ObjectName bean = _beans.remove(obj);
if (bean != null)
@@ -202,7 +202,7 @@ public class MBeanContainer extends AbstractLifeCycle implements Container.Liste
List beanRelations= _relations.remove(bean);
if (beanRelations != null)
{
- __log.debug("Unregister {}", beanRelations);
+ LOG.debug("Unregister {}", beanRelations);
List removeList = new ArrayList