refactored client to use upgradeable endpoint. Instert SslConnection when needed

jetty-client/src/main/java/org/eclipse/jetty/client/AbstractHttpConnection.java

@@ -13,7 +13,6 @@
 
 package org.eclipse.jetty.client;
 
-import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Collections;
@@ -354,7 +353,8 @@ public abstract class AbstractHttpConnection extends AbstractConnection implemen
     @Override
     public String toString()
     {
-        return "HttpConnection@" + hashCode() + "//" + _destination.getAddress().getHost() + ":" + _destination.getAddress().getPort();
+        return "HttpConnection@" + hashCode() + "//" + 
+        (_destination==null?"?.?.?.?:??":(_destination.getAddress().getHost() + ":" + _destination.getAddress().getPort()));
     }
 
     public String toDetailString()

jetty-client/src/main/java/org/eclipse/jetty/client/HttpDestination.java

@@ -352,9 +352,9 @@ public class HttpDestination implements Dumpable
             else
             {
                 EndPoint endPoint = connection.getEndPoint();
-                if (isProxied() && endPoint instanceof SelectConnector.ProxySelectChannelEndPoint)
+                if (isProxied() && endPoint instanceof SelectConnector.UpgradableEndPoint)
                 {
-                    SelectConnector.ProxySelectChannelEndPoint proxyEndPoint = (SelectConnector.ProxySelectChannelEndPoint)endPoint;
+                    SelectConnector.UpgradableEndPoint proxyEndPoint = (SelectConnector.UpgradableEndPoint)endPoint;
                     HttpExchange exchange = _queue.get(0);
                     ConnectExchange connect = new ConnectExchange(getAddress(), proxyEndPoint, exchange);
                     connect.setAddress(getProxy());
@@ -668,10 +668,10 @@ public class HttpDestination implements Dumpable
     
     private class ConnectExchange extends ContentExchange
     {
-        private final SelectConnector.ProxySelectChannelEndPoint proxyEndPoint;
+        private final SelectConnector.UpgradableEndPoint proxyEndPoint;
         private final HttpExchange exchange;
 
-        public ConnectExchange(Address serverAddress, SelectConnector.ProxySelectChannelEndPoint proxyEndPoint, HttpExchange exchange)
+        public ConnectExchange(Address serverAddress, SelectConnector.UpgradableEndPoint proxyEndPoint, HttpExchange exchange)
         {
             this.proxyEndPoint = proxyEndPoint;
             this.exchange = exchange;

jetty-client/src/main/java/org/eclipse/jetty/client/SelectConnector.java

@@ -24,23 +24,23 @@ import java.util.concurrent.ConcurrentHashMap;
 import javax.net.ssl.SSLEngine;
 import javax.net.ssl.SSLSession;
 
-import org.eclipse.jetty.http.HttpGenerator;
+import org.eclipse.jetty.io.AsyncEndPoint;
-import org.eclipse.jetty.http.HttpParser;
 import org.eclipse.jetty.io.Buffer;
 import org.eclipse.jetty.io.Buffers;
 import org.eclipse.jetty.io.Buffers.Type;
 import org.eclipse.jetty.io.BuffersFactory;
 import org.eclipse.jetty.io.ConnectedEndPoint;
 import org.eclipse.jetty.io.Connection;
-import org.eclipse.jetty.io.EndPoint;
 import org.eclipse.jetty.io.nio.AsyncConnection;
 import org.eclipse.jetty.io.nio.SelectChannelEndPoint;
 import org.eclipse.jetty.io.nio.SelectorManager;
+import org.eclipse.jetty.io.nio.SslConnection;
 import org.eclipse.jetty.util.component.AbstractLifeCycle;
 import org.eclipse.jetty.util.log.Log;
 import org.eclipse.jetty.util.log.Logger;
 import org.eclipse.jetty.util.ssl.SslContextFactory;
 import org.eclipse.jetty.util.thread.Timeout;
+import org.eclipse.jetty.util.thread.Timeout.Task;
 
 class SelectConnector extends AbstractLifeCycle implements HttpClient.Connector
 {
@@ -49,8 +49,7 @@ class SelectConnector extends AbstractLifeCycle implements HttpClient.Connector
     private final HttpClient _httpClient;
     private final Manager _selectorManager=new Manager();
     private final Map<SocketChannel, Timeout.Task> _connectingChannels = new ConcurrentHashMap<SocketChannel, Timeout.Task>();
-    private Buffers _sslBuffers;
+    
-
     /**
      * @param httpClient the HttpClient this connector is associated to
      */
@@ -65,16 +64,6 @@ class SelectConnector extends AbstractLifeCycle implements HttpClient.Connector
     {
         super.doStart();
 
-
-        final boolean direct=_httpClient.getUseDirectBuffers();
-
-        SSLEngine sslEngine=_selectorManager.newSslEngine(null);
-        final SSLSession ssl_session=sslEngine.getSession();
-        _sslBuffers = BuffersFactory.newBuffers(
-                direct?Type.DIRECT:Type.INDIRECT,ssl_session.getApplicationBufferSize(),
-                direct?Type.DIRECT:Type.INDIRECT,ssl_session.getApplicationBufferSize(),
-                direct?Type.DIRECT:Type.INDIRECT,1024);
-
         _selectorManager.start();
     }
 
@@ -129,6 +118,8 @@ class SelectConnector extends AbstractLifeCycle implements HttpClient.Connector
     /* ------------------------------------------------------------ */
     class Manager extends SelectorManager
     {
+        Logger LOG = SelectConnector.LOG;
+        
         @Override
         public boolean dispatch(Runnable task)
         {
@@ -151,11 +142,8 @@ class SelectConnector extends AbstractLifeCycle implements HttpClient.Connector
         }
 
         @Override
-        protected AsyncConnection newConnection(SocketChannel channel, SelectChannelEndPoint endpoint)
+        public AsyncConnection newConnection(SocketChannel channel, AsyncEndPoint endpoint, Object attachment)
         {
-            if (endpoint instanceof SslSelectChannelEndPoint)
-                return new AsyncHttpConnection(_sslBuffers,_sslBuffers,endpoint);
-
             return new AsyncHttpConnection(_httpClient.getRequestBuffers(),_httpClient.getResponseBuffers(),endpoint);
         }
 
@@ -172,32 +160,29 @@ class SelectConnector extends AbstractLifeCycle implements HttpClient.Connector
             // key should have destination at this point (will be replaced by endpoint after this call)
             HttpDestination dest=(HttpDestination)key.attachment();
 
-            SelectChannelEndPoint ep=null;
+            AsyncEndPoint ep=null;
 
+            SelectChannelEndPoint scep= new SelectChannelEndPoint(channel, selectSet, key, (int)_httpClient.getIdleTimeout());
+            ep = scep;
+            
             if (dest.isSecure())
             {
-                if (dest.isProxied())
+                LOG.debug("secure to {}, proxied={}",channel,dest.isProxied());
-                {
+                ep = new UpgradableEndPoint(ep,newSslEngine(channel));
-                    SSLEngine engine=newSslEngine(channel);
-                    ep = new ProxySelectChannelEndPoint(channel, selectSet, key, _sslBuffers, engine, (int)_httpClient.getIdleTimeout());
-                }
-                else
-                {
-                    SSLEngine engine=newSslEngine(channel);
-                    SslSelectChannelEndPoint sslEp = new SslSelectChannelEndPoint(_sslBuffers, channel, selectSet, key, engine, (int)_httpClient.getIdleTimeout());
-                    sslEp.setAllowRenegotiate(_httpClient.getSslContextFactory().isAllowRenegotiate());
-                    ep = sslEp;
-                }
             }
-            else
+            
-            {
+            AsyncConnection connection = selectSet.getManager().newConnection(channel,ep, key.attachment());
-                ep = new SelectChannelEndPoint(channel, selectSet, key, (int)_httpClient.getIdleTimeout());
+            ep.setConnection(connection);
-            }
+            
-
+            AbstractHttpConnection httpConnection=(AbstractHttpConnection)connection;
-            AbstractHttpConnection connection=(AbstractHttpConnection)ep.getConnection();
+            httpConnection.setDestination(dest);
-            connection.setDestination(dest);
+            
-            dest.onNewConnection(connection);
+            if (dest.isSecure() && !dest.isProxied())
-            return ep;
+                ((UpgradableEndPoint)ep).upgrade();
+            
+            dest.onNewConnection(httpConnection);
+            
+            return scep;
         }
 
         private synchronized SSLEngine newSslEngine(SocketChannel channel) throws IOException
@@ -268,204 +253,206 @@ class SelectConnector extends AbstractLifeCycle implements HttpClient.Connector
             }
         }
     }
-
+    
-    /**
+    public static class UpgradableEndPoint implements AsyncEndPoint
-     * An endpoint that is able to "upgrade" from a normal endpoint to a SSL endpoint.
-     * Since {@link HttpParser} and {@link HttpGenerator} only depend on the {@link EndPoint}
-     * interface, this class overrides all methods of {@link EndPoint} to provide the right
-     * behavior depending on the fact that it has been upgraded or not.
-     */
-    public static class ProxySelectChannelEndPoint extends SslSelectChannelEndPoint
     {
-        private final SelectChannelEndPoint plainEndPoint;
+        AsyncEndPoint _endp;
-        private volatile boolean upgraded = false;
+        SSLEngine _engine;
-
+        
-        public ProxySelectChannelEndPoint(SocketChannel channel, SelectorManager.SelectSet selectSet, SelectionKey key, Buffers sslBuffers, SSLEngine engine, int maxIdleTimeout) throws IOException
+        public UpgradableEndPoint(AsyncEndPoint endp, SSLEngine engine) throws IOException
         {
-            super(sslBuffers, channel, selectSet, key, engine, maxIdleTimeout);
+            _engine=engine;
-            this.plainEndPoint = new SelectChannelEndPoint(channel, selectSet, key, maxIdleTimeout);
+            _endp=endp;
         }
 
         public void upgrade()
         {
-            upgraded = true;
+            AsyncHttpConnection connection = (AsyncHttpConnection) ((SelectChannelEndPoint)_endp).getConnection();
+            
+            SslConnection sslConnection = new SslConnection(_engine,_endp);
+            ((SelectChannelEndPoint)_endp).setConnection(sslConnection);
+
+            _endp=sslConnection.getSslEndPoint();
+            sslConnection.setConnection(connection);
+            
+            LOG.debug("upgrade {} to {} for {}",this,sslConnection,connection);
+        }
+        
+        
+        public Connection getConnection()
+        {
+            return _endp.getConnection();
+        }
+
+        public void setConnection(Connection connection)
+        {
+            _endp.setConnection(connection);
         }
 
         public void shutdownOutput() throws IOException
         {
-            if (upgraded)
+            _endp.shutdownOutput();
-                super.shutdownOutput();
+        }
-            else
+
-                plainEndPoint.shutdownOutput();
+        public void asyncDispatch()
+        {
+            _endp.asyncDispatch();
+        }
+
+        public boolean isOutputShutdown()
+        {
+            return _endp.isOutputShutdown();
+        }
+
+        public void shutdownInput() throws IOException
+        {
+            _endp.shutdownInput();
+        }
+
+        public void scheduleWrite()
+        {
+            _endp.scheduleWrite();
+        }
+
+        public boolean isInputShutdown()
+        {
+            return _endp.isInputShutdown();
         }
 
         public void close() throws IOException
         {
-            if (upgraded)
+            _endp.close();
-                super.close();
+        }
-            else
+
-                plainEndPoint.close();
+        public void scheduleIdle()
+        {
+            _endp.scheduleIdle();
         }
 
         public int fill(Buffer buffer) throws IOException
         {
-            if (upgraded)
+            return _endp.fill(buffer);
-                return super.fill(buffer);
+        }
-            else
+
-                return plainEndPoint.fill(buffer);
+        public void cancelIdle()
+        {
+            _endp.cancelIdle();
+        }
+
+        public boolean isWritable()
+        {
+            return _endp.isWritable();
+        }
+
+        public boolean hasProgressed()
+        {
+            return _endp.hasProgressed();
         }
 
         public int flush(Buffer buffer) throws IOException
         {
-            if (upgraded)
+            return _endp.flush(buffer);
-                return super.flush(buffer);
+        }
-            else
+
-                return plainEndPoint.flush(buffer);
+        public void scheduleTimeout(Task task, long timeoutMs)
+        {
+            _endp.scheduleTimeout(task,timeoutMs);
+        }
+
+        public void cancelTimeout(Task task)
+        {
+            _endp.cancelTimeout(task);
         }
 
         public int flush(Buffer header, Buffer buffer, Buffer trailer) throws IOException
         {
-            if (upgraded)
+            return _endp.flush(header,buffer,trailer);
-                return super.flush(header, buffer, trailer);
-            else
-                return plainEndPoint.flush(header, buffer, trailer);
         }
 
         public String getLocalAddr()
         {
-            if (upgraded)
+            return _endp.getLocalAddr();
-                return super.getLocalAddr();
-            else
-                return plainEndPoint.getLocalAddr();
         }
 
         public String getLocalHost()
         {
-            if (upgraded)
+            return _endp.getLocalHost();
-                return super.getLocalHost();
-            else
-                return plainEndPoint.getLocalHost();
         }
 
         public int getLocalPort()
         {
-            if (upgraded)
+            return _endp.getLocalPort();
-                return super.getLocalPort();
-            else
-                return plainEndPoint.getLocalPort();
         }
 
         public String getRemoteAddr()
         {
-            if (upgraded)
+            return _endp.getRemoteAddr();
-                return super.getRemoteAddr();
-            else
-                return plainEndPoint.getRemoteAddr();
         }
 
         public String getRemoteHost()
         {
-            if (upgraded)
+            return _endp.getRemoteHost();
-                return super.getRemoteHost();
-            else
-                return plainEndPoint.getRemoteHost();
         }
 
         public int getRemotePort()
         {
-            if (upgraded)
+            return _endp.getRemotePort();
-                return super.getRemotePort();
-            else
-                return plainEndPoint.getRemotePort();
         }
 
         public boolean isBlocking()
         {
-            if (upgraded)
+            return _endp.isBlocking();
-                return super.isBlocking();
-            else
-                return plainEndPoint.isBlocking();
         }
 
         public boolean isBufferred()
         {
-            if (upgraded)
+            return _endp.isBufferred();
-                return super.isBufferred();
-            else
-                return plainEndPoint.isBufferred();
         }
 
         public boolean blockReadable(long millisecs) throws IOException
         {
-            if (upgraded)
+            return _endp.blockReadable(millisecs);
-                return super.blockReadable(millisecs);
-            else
-                return plainEndPoint.blockReadable(millisecs);
         }
 
         public boolean blockWritable(long millisecs) throws IOException
         {
-            if (upgraded)
+            return _endp.blockWritable(millisecs);
-                return super.blockWritable(millisecs);
-            else
-                return plainEndPoint.blockWritable(millisecs);
         }
 
         public boolean isOpen()
         {
-            if (upgraded)
+            return _endp.isOpen();
-                return super.isOpen();
-            else
-                return plainEndPoint.isOpen();
         }
 
         public Object getTransport()
         {
-            if (upgraded)
+            return _endp.getTransport();
-                return super.getTransport();
-            else
-                return plainEndPoint.getTransport();
         }
 
         public boolean isBufferingInput()
         {
-            if (upgraded)
+            return _endp.isBufferingInput();
-                return super.isBufferingInput();
-            else
-                return plainEndPoint.isBufferingInput();
         }
 
         public boolean isBufferingOutput()
         {
-            if (upgraded)
+            return _endp.isBufferingOutput();
-                return super.isBufferingOutput();
-            else
-                return plainEndPoint.isBufferingOutput();
         }
 
         public void flush() throws IOException
         {
-            if (upgraded)
+            _endp.flush();
-                super.flush();
-            else
-                plainEndPoint.flush();
-
         }
 
         public int getMaxIdleTime()
         {
-            if (upgraded)
+            return _endp.getMaxIdleTime();
-                return super.getMaxIdleTime();
-            else
-                return plainEndPoint.getMaxIdleTime();
         }
 
         public void setMaxIdleTime(int timeMs) throws IOException
         {
-            if (upgraded)
+            _endp.setMaxIdleTime(timeMs);
-                super.setMaxIdleTime(timeMs);
-            else
-                plainEndPoint.setMaxIdleTime(timeMs);
         }
+
     }
 }

jetty-client/src/test/java/org/eclipse/jetty/client/AsyncSslHttpExchangeTest.java

@@ -30,11 +30,11 @@ public class AsyncSslHttpExchangeTest extends SslHttpExchangeTest
         _port = _server.getConnectors()[0].getLocalPort();
     }
 
-    @Override
-    public void testPerf() throws Exception
-    {
-        sender(10,true);
-    }
 
+    @Override
+    public void testGetWithContentExchange() throws Exception
+    {
+        super.testGetWithContentExchange();
+    }
     
 }

jetty-io/src/main/java/org/eclipse/jetty/io/AbstractConnection.java

@@ -11,7 +11,7 @@ public abstract class AbstractConnection implements Connection
     private static final Logger LOG = Log.getLogger(AbstractConnection.class);
 
     private final long _timeStamp;
-    public final EndPoint _endp; // TODO make private
+    protected final EndPoint _endp; 
 
     public AbstractConnection(EndPoint endp)
     {
@@ -59,6 +59,6 @@ public abstract class AbstractConnection implements Connection
     
     public String toString()
     {
-        return super.toString()+"@"+_endp.getLocalAddr()+":"+_endp.getLocalPort()+"<->"+_endp.getRemoteAddr()+":"+_endp.getRemotePort();
+        return this.getClass().getSimpleName()+"@"+_endp.getLocalAddr()+":"+_endp.getLocalPort()+"<->"+_endp.getRemoteAddr()+":"+_endp.getRemotePort();
     }
 }

jetty-io/src/main/java/org/eclipse/jetty/io/AsyncEndPoint.java

@@ -51,6 +51,13 @@ public interface AsyncEndPoint extends EndPoint
     public boolean hasProgressed();
     
 
+    /* ------------------------------------------------------------ */
+    public Connection getConnection();
+
+    /* ------------------------------------------------------------ */
+    public void setConnection(Connection connection);
+    
+    
     /* ------------------------------------------------------------ */
     /**
      */

jetty-io/src/main/java/org/eclipse/jetty/io/nio/SelectChannelEndPoint.java

@@ -94,10 +94,7 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
         _open=true;
         _key = key;
 
-        _connection = _manager.newConnection(channel,this);
-
         scheduleIdle();
-        
     }
     
     /* ------------------------------------------------------------ */
@@ -124,10 +121,10 @@ public class SelectChannelEndPoint extends ChannelEndPoint implements AsyncEndPo
     /* ------------------------------------------------------------ */
     public void setConnection(Connection connection)
     {
-        // TODO Only needed for local connection
         Connection old=_connection;
         _connection=(AsyncConnection)connection;
-        _manager.endPointUpgraded(this,old);
+        if (old!=null && old!=_connection)
+            _manager.endPointUpgraded(this,old);
     }
 
     /* ------------------------------------------------------------ */

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;
@@ -337,7 +338,7 @@ public abstract class SelectorManager extends AbstractLifeCycle implements Dumpa
     protected abstract void endPointUpgraded(ConnectedEndPoint endpoint,Connection oldConnection);
 
     /* ------------------------------------------------------------------------------- */
-    protected abstract AsyncConnection newConnection(SocketChannel channel, SelectChannelEndPoint endpoint);
+    public abstract AsyncConnection newConnection(SocketChannel channel, AsyncEndPoint endpoint, Object attachment);
 
     /* ------------------------------------------------------------ */
     /**

jetty-io/src/main/java/org/eclipse/jetty/io/nio/SslConnection.java

@@ -59,12 +59,14 @@ public class SslConnection extends AbstractConnection implements AsyncConnection
     private boolean _allowRenegotiate=true;
     private boolean _handshook;
 
-    
+
+    /* ------------------------------------------------------------ */
     public SslConnection(SSLEngine engine,EndPoint endp)
     {
         this(engine,endp,System.currentTimeMillis());
     }
-    
+
+    /* ------------------------------------------------------------ */
     public SslConnection(SSLEngine engine,EndPoint endp, long timeStamp)
     {
         super(endp,timeStamp);
@@ -72,12 +74,14 @@ public class SslConnection extends AbstractConnection implements AsyncConnection
         _session=_engine.getSession();
         _aEndp=(AsyncEndPoint)endp;
     }
-    
+
+    /* ------------------------------------------------------------ */
     public synchronized void setConnection(AsyncConnection connection)
     {
         _connection=connection;
     }
-    
+
+    /* ------------------------------------------------------------ */
     public synchronized AsyncConnection getConnection()
     {
         return _connection;
@@ -478,8 +482,8 @@ public class SslConnection extends AbstractConnection implements AsyncConnection
             throw new IOException(result.toString());
         }
         
-        if (LOG.isDebugEnabled() && result.bytesProduced()>0)
+        //if (LOG.isDebugEnabled() && result.bytesProduced()>0)
-            LOG.debug("{} unwrapped '{}'",_session,buffer);
+        //    LOG.debug("{} unwrapped '{}'",_session,buffer);
         
         return result.bytesConsumed()>0 || result.bytesProduced()>0;
     }
@@ -721,6 +725,16 @@ public class SslConnection extends AbstractConnection implements AsyncConnection
             _aEndp.setMaxIdleTime(timeMs);
         }
 
+        public Connection getConnection()
+        {
+            return _connection;
+        }
+
+        public void setConnection(Connection connection)
+        {
+            _connection=(AsyncConnection)connection;            
+        }
+
         
     }
     

jetty-io/src/test/java/org/eclipse/jetty/io/nio/SelectChannelEndPointTest.java

@@ -11,6 +11,7 @@ import java.nio.channels.ServerSocketChannel;
 import java.nio.channels.SocketChannel;
 
 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;
@@ -48,15 +49,17 @@ public class SelectChannelEndPointTest
         }
 
         @Override
-        protected AsyncConnection newConnection(SocketChannel channel, SelectChannelEndPoint endpoint)
+        public AsyncConnection newConnection(SocketChannel channel, AsyncEndPoint endpoint, Object attachment)
         {
             return SelectChannelEndPointTest.this.newConnection(channel,endpoint);
         }
 
         @Override
-        protected SelectChannelEndPoint newEndPoint(SocketChannel channel, SelectSet selectSet, SelectionKey sKey) throws IOException
+        protected SelectChannelEndPoint newEndPoint(SocketChannel channel, SelectSet selectSet, SelectionKey key) throws IOException
         {
-            return new SelectChannelEndPoint(channel,selectSet,sKey,2000);
+            SelectChannelEndPoint endp = new SelectChannelEndPoint(channel,selectSet,key,2000);
+            endp.setConnection(selectSet.getManager().newConnection(channel,endp, key.attachment()));
+            return endp;
         }    
     };
 

jetty-server/src/main/java/org/eclipse/jetty/server/AsyncHttpConnection.java

@@ -26,7 +26,7 @@ public class AsyncHttpConnection extends AbstractHttpConnection implements Async
         super(connector,endpoint,server);
         _asyncEndp=(AsyncEndPoint)endpoint;
     }
-
+    
     public Connection handle() throws IOException
     {
         Connection connection = this;

jetty-server/src/main/java/org/eclipse/jetty/server/LocalConnector.java

@@ -110,7 +110,8 @@ public class LocalConnector extends AbstractConnector
                     @Override
                     public void setConnection(Connection connection)
                     {
-                        connectionUpgraded(getConnection(),connection);
+                        if (getConnection()!=null && connection!=getConnection())
+                            connectionUpgraded(getConnection(),connection);
                         super.setConnection(connection);
                     }
                 };

jetty-server/src/main/java/org/eclipse/jetty/server/bio/SocketConnector.java

@@ -193,7 +193,7 @@ public class SocketConnector extends AbstractConnector
 
         public void setConnection(Connection connection)
         {
-            if (_connection!=connection)
+            if (_connection!=connection && _connection!=null)
                 connectionUpgraded(_connection,connection);
             _connection=connection;
         }

jetty-server/src/main/java/org/eclipse/jetty/server/handler/ConnectHandler.java

@@ -17,6 +17,7 @@ import javax.servlet.http.HttpServletResponse;
 
 import org.eclipse.jetty.http.HttpMethods;
 import org.eclipse.jetty.http.HttpParser;
+import org.eclipse.jetty.io.AsyncEndPoint;
 import org.eclipse.jetty.io.Buffer;
 import org.eclipse.jetty.io.ConnectedEndPoint;
 import org.eclipse.jetty.io.Connection;
@@ -422,17 +423,18 @@ public class ConnectHandler extends HandlerWrapper
     private class Manager extends SelectorManager
     {
         @Override
-        protected SelectChannelEndPoint newEndPoint(SocketChannel channel, SelectSet selectSet, SelectionKey selectionKey) throws IOException
+        protected SelectChannelEndPoint newEndPoint(SocketChannel channel, SelectSet selectSet, SelectionKey key) throws IOException
         {
-            SelectChannelEndPoint endp = new SelectChannelEndPoint(channel, selectSet, selectionKey, channel.socket().getSoTimeout());
+            SelectChannelEndPoint endp = new SelectChannelEndPoint(channel, selectSet, key, channel.socket().getSoTimeout());
+            endp.setConnection(selectSet.getManager().newConnection(channel,endp, key.attachment()));
             endp.setMaxIdleTime(_writeTimeout);
             return endp;
         }
 
         @Override
-        protected AsyncConnection newConnection(SocketChannel channel, SelectChannelEndPoint endpoint)
+        public AsyncConnection newConnection(SocketChannel channel, AsyncEndPoint endpoint, Object attachment)
         {
-            ProxyToServerConnection proxyToServer = (ProxyToServerConnection)endpoint.getSelectionKey().attachment();
+            ProxyToServerConnection proxyToServer = (ProxyToServerConnection)attachment;
             proxyToServer.setTimeStamp(System.currentTimeMillis());
             proxyToServer.setEndPoint(endpoint);
             return proxyToServer;
@@ -472,7 +474,7 @@ public class ConnectHandler extends HandlerWrapper
         private volatile Buffer _data;
         private volatile ClientToProxyConnection _toClient;
         private volatile long _timestamp;
-        private volatile SelectChannelEndPoint _endPoint;
+        private volatile AsyncEndPoint _endPoint;
 
         public ProxyToServerConnection(ConcurrentMap<String, Object> context, Buffer data)
         {
@@ -589,7 +591,7 @@ public class ConnectHandler extends HandlerWrapper
             _timestamp = timestamp;
         }
 
-        public void setEndPoint(SelectChannelEndPoint endpoint)
+        public void setEndPoint(AsyncEndPoint endpoint)
         {
             _endPoint = endpoint;
         }

jetty-server/src/main/java/org/eclipse/jetty/server/nio/NetworkTrafficSelectChannelConnector.java

@@ -54,6 +54,7 @@ public class NetworkTrafficSelectChannelConnector extends SelectChannelConnector
     protected SelectChannelEndPoint newEndPoint(SocketChannel channel, SelectorManager.SelectSet selectSet, SelectionKey key) throws IOException
     {
         NetworkTrafficSelectChannelEndPoint endPoint = new NetworkTrafficSelectChannelEndPoint(channel, selectSet, key, _maxIdleTime, listeners);
+        endPoint.setConnection(selectSet.getManager().newConnection(channel,endPoint, key.attachment()));
         endPoint.notifyOpened();
         return endPoint;
     }

jetty-server/src/main/java/org/eclipse/jetty/server/nio/SelectChannelConnector.java

@@ -278,7 +278,9 @@ public class SelectChannelConnector extends AbstractNIOConnector
     /* ------------------------------------------------------------ */
     protected SelectChannelEndPoint newEndPoint(SocketChannel channel, SelectSet selectSet, SelectionKey key) throws IOException
     {
-        return new SelectChannelEndPoint(channel,selectSet,key, SelectChannelConnector.this._maxIdleTime);
+        SelectChannelEndPoint endp= new SelectChannelEndPoint(channel,selectSet,key, SelectChannelConnector.this._maxIdleTime);
+        endp.setConnection(selectSet.getManager().newConnection(channel,endp, key.attachment()));
+        return endp;
     }
 
     /* ------------------------------------------------------------------------------- */
@@ -353,7 +355,7 @@ public class SelectChannelConnector extends AbstractNIOConnector
         }
 
         @Override
-        protected AsyncConnection newConnection(SocketChannel channel,SelectChannelEndPoint endpoint)
+        public AsyncConnection newConnection(SocketChannel channel,AsyncEndPoint endpoint, Object attachment)
         {
             return SelectChannelConnector.this.newConnection(channel,endpoint);
         }

jetty-server/src/test/java/org/eclipse/jetty/server/BusySelectChannelServerTest.java

@@ -16,8 +16,10 @@ import java.io.IOException;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.SocketChannel;
 
+import org.eclipse.jetty.io.AsyncEndPoint;
 import org.eclipse.jetty.io.Buffer;
 import org.eclipse.jetty.io.View;
+import org.eclipse.jetty.io.nio.AsyncConnection;
 import org.eclipse.jetty.io.nio.IndirectNIOBuffer;
 import org.eclipse.jetty.io.nio.NIOBuffer;
 import org.eclipse.jetty.io.nio.SelectChannelEndPoint;
@@ -38,7 +40,7 @@ public class BusySelectChannelServerTest extends HttpServerTestBase
             @Override
             protected SelectChannelEndPoint newEndPoint(SocketChannel channel, SelectSet selectSet, SelectionKey key) throws IOException
             {
-                return new SelectChannelEndPoint(channel,selectSet,key, _maxIdleTime)
+                SelectChannelEndPoint endp=new SelectChannelEndPoint(channel,selectSet,key, _maxIdleTime)
                 {
                     int write;
                     int read;
@@ -134,6 +136,8 @@ public class BusySelectChannelServerTest extends HttpServerTestBase
                         return super.fill(buffer);
                     }
                 };
+                endp.setConnection(selectSet.getManager().newConnection(channel,endp, key.attachment()));
+                return endp;
             }
         };
         connector.setAcceptors(1);

jetty-websocket/src/main/java/org/eclipse/jetty/websocket/WebSocketClientFactory.java

@@ -10,6 +10,7 @@ import java.util.Random;
 import org.eclipse.jetty.http.HttpFields;
 import org.eclipse.jetty.http.HttpParser;
 import org.eclipse.jetty.io.AbstractConnection;
+import org.eclipse.jetty.io.AsyncEndPoint;
 import org.eclipse.jetty.io.Buffer;
 import org.eclipse.jetty.io.Buffers;
 import org.eclipse.jetty.io.ByteArrayBuffer;
@@ -206,15 +207,17 @@ public class WebSocketClientFactory extends AggregateLifeCycle
         }
 
         @Override
-        protected SelectChannelEndPoint newEndPoint(SocketChannel channel, SelectSet selectSet, final SelectionKey sKey) throws IOException
+        protected SelectChannelEndPoint newEndPoint(SocketChannel channel, SelectSet selectSet, final SelectionKey key) throws IOException
         {
-            return new SelectChannelEndPoint(channel,selectSet,sKey,channel.socket().getSoTimeout());
+            SelectChannelEndPoint endp= new SelectChannelEndPoint(channel,selectSet,key,channel.socket().getSoTimeout());
+            endp.setConnection(selectSet.getManager().newConnection(channel,endp, key.attachment()));
+            return endp;
         }
 
         @Override
-        protected AsyncConnection newConnection(SocketChannel channel, SelectChannelEndPoint endpoint)
+        public AsyncConnection newConnection(SocketChannel channel, AsyncEndPoint endpoint, Object attachment)
         {
-            WebSocketClient.WebSocketFuture holder = (WebSocketClient.WebSocketFuture) endpoint.getSelectionKey().attachment();
+            WebSocketClient.WebSocketFuture holder = (WebSocketClient.WebSocketFuture) attachment;
             return new HandshakeConnection(endpoint,holder);
         }
 
@@ -258,14 +261,14 @@ public class WebSocketClientFactory extends AggregateLifeCycle
      */
     class HandshakeConnection extends AbstractConnection implements AsyncConnection
     {
-        private final SelectChannelEndPoint _endp;
+        private final AsyncEndPoint _endp;
         private final WebSocketClient.WebSocketFuture _future;
         private final String _key;
         private final HttpParser _parser;
         private String _accept;
         private String _error;
 
-        public HandshakeConnection(SelectChannelEndPoint endpoint, WebSocketClient.WebSocketFuture future)
+        public HandshakeConnection(AsyncEndPoint endpoint, WebSocketClient.WebSocketFuture future)
         {
             super(endpoint,System.currentTimeMillis());
             _endp=endpoint;