Issue #2014 - Unix Socket Client

jetty-client/pom.xml

@@ -105,7 +105,6 @@
       <version>${project.version}</version>
       <optional>true</optional>
     </dependency>
-
     <dependency>
       <groupId>org.eclipse.jetty</groupId>
       <artifactId>jetty-server</artifactId>

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

@@ -753,7 +753,7 @@ public void run()
             List<EndPoint> end_points = new ArrayList<>();
             for (SelectionKey key : _selector.keys())
             {
-                if (key.isValid())
+                if (key!=null && key.isValid())
                 {
                     Object attachment = key.attachment();
                     if (attachment instanceof EndPoint)

jetty-unixsocket/pom.xml

@@ -32,7 +32,12 @@
     <dependency>
       <groupId>com.github.jnr</groupId>
       <artifactId>jnr-unixsocket</artifactId>
-      <version>0.18</version>
+    </dependency>
+    <dependency>
+      <groupId>org.eclipse.jetty</groupId>
+      <artifactId>jetty-client</artifactId>
+      <version>${project.version}</version>
+      <scope>provided</scope>
     </dependency>
     <dependency>
       <groupId>org.eclipse.jetty.toolchain</groupId>

jetty-unixsocket/src/main/java/org/eclipse/jetty/unixsocket/UnixSocketConnector.java

@@ -248,8 +248,8 @@ public void open() throws IOException
             serverChannel.configureBlocking(getAcceptors()>0);
             serverChannel.socket().bind(bindAddress, getAcceptQueueSize());
             addBean(serverChannel);
-
-            LOG.debug("opened {}",serverChannel);
+            if (LOG.isDebugEnabled())
+                LOG.debug("opened {}",serverChannel);
             _acceptChannel = serverChannel;
         }
     }
@@ -430,9 +430,11 @@ protected boolean isConnectionPending(SelectableChannel channel)
         @Override
         protected SelectableChannel doAccept(SelectableChannel server) throws IOException
         {
-            LOG.debug("doAccept async {}",server);
+            if (LOG.isDebugEnabled())
+                LOG.debug("doAccept async {}",server);
             UnixSocketChannel channel = ((UnixServerSocketChannel)server).accept();
-            LOG.debug("accepted async {}",channel);
+            if (LOG.isDebugEnabled())
+                LOG.debug("accepted async {}",channel);
             return channel;
         }
     }

jetty-unixsocket/src/main/java/org/eclipse/jetty/unixsocket/UnixSocketEndPoint.java

@@ -20,10 +20,13 @@
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
 import java.nio.channels.SelectionKey;
 
 import org.eclipse.jetty.io.ChannelEndPoint;
+import org.eclipse.jetty.io.EofException;
 import org.eclipse.jetty.io.ManagedSelector;
+import org.eclipse.jetty.util.BufferUtil;
 import org.eclipse.jetty.util.log.Log;
 import org.eclipse.jetty.util.log.Logger;
 import org.eclipse.jetty.util.thread.Scheduler;
@@ -32,8 +35,9 @@
 
 public class UnixSocketEndPoint extends ChannelEndPoint
 {
-    public final static InetSocketAddress NOIP=new InetSocketAddress(0);
     private static final Logger LOG = Log.getLogger(UnixSocketEndPoint.class);
+    private static final Logger CEPLOG = Log.getLogger(ChannelEndPoint.class);
+
 
     private final UnixSocketChannel _channel;
 
@@ -71,4 +75,50 @@ protected void doShutdownOutput()
             LOG.debug(e);
         }
     }
+
+
+    @Override
+    public boolean flush(ByteBuffer... buffers) throws IOException
+    {
+        // TODO this is a work around for https://github.com/jnr/jnr-unixsocket/issues/50
+        long flushed=0;
+        try
+        {
+            for (ByteBuffer b : buffers)
+            {
+                if (b.hasRemaining())
+                {
+                    int r=b.remaining();
+                    int p=b.position();
+                    int l=_channel.write(b);
+                    if (l>=0)
+                    {
+                        b.position(p+l);
+                        flushed+=l;
+                    }
+
+                    if (CEPLOG.isDebugEnabled())
+                        CEPLOG.debug("flushed {}/{} r={} {}", l,r,b.remaining(), this);
+
+                    if (b.hasRemaining())
+                        break;
+                }
+            }
+
+        }
+        catch (IOException e)
+        {
+            throw new EofException(e);
+        }
+
+        if (flushed>0)
+            notIdle();
+
+        for (ByteBuffer b : buffers)
+            if (!BufferUtil.isEmpty(b))
+                return false;
+
+        return true;
+    }
+
 }

jetty-unixsocket/src/main/java/org/eclipse/jetty/unixsocket/client/HttpClientTransportOverUnixSockets.java

@@ -0,0 +1,141 @@
+//
+//  ========================================================================
+//  Copyright (c) 1995-2017 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.unixsocket.client;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.SocketException;
+import java.nio.channels.SelectableChannel;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.util.Map;
+
+import org.eclipse.jetty.client.HttpClient;
+import org.eclipse.jetty.client.HttpDestination;
+import org.eclipse.jetty.client.http.HttpClientTransportOverHTTP;
+import org.eclipse.jetty.io.EndPoint;
+import org.eclipse.jetty.io.ManagedSelector;
+import org.eclipse.jetty.io.SelectorManager;
+import org.eclipse.jetty.io.ssl.SslClientConnectionFactory;
+import org.eclipse.jetty.unixsocket.UnixSocketEndPoint;
+import org.eclipse.jetty.util.log.Log;
+import org.eclipse.jetty.util.log.Logger;
+
+import jnr.enxio.channels.NativeSelectorProvider;
+import jnr.unixsocket.UnixSocketAddress;
+import jnr.unixsocket.UnixSocketChannel;
+
+public class HttpClientTransportOverUnixSockets
+    extends HttpClientTransportOverHTTP
+{
+    private static final Logger LOG = Log.getLogger( HttpClientTransportOverUnixSockets.class );
+
+    private String _unixSocket;
+    private SelectorManager selectorManager;
+
+
+    public HttpClientTransportOverUnixSockets( String unixSocket )
+    {
+        if ( unixSocket == null )
+        {
+            throw new IllegalArgumentException( "Unix socket file cannot be null" );
+        }
+        this._unixSocket = unixSocket;
+    }
+
+    @Override
+    protected SelectorManager newSelectorManager(HttpClient client)
+    {
+        return selectorManager = new UnixSocketSelectorManager(client,getSelectors());
+    }
+
+    @Override
+    public void connect( InetSocketAddress address, Map<String, Object> context )
+    {
+        UnixSocketChannel channel = null;
+        try
+        {
+            InetAddress inet = address.getAddress();
+            if (!inet.isLoopbackAddress() && !inet.isLinkLocalAddress() && !inet.isSiteLocalAddress())
+                throw new IOException("UnixSocket cannot connect to "+address.getHostString());
+
+            // Open a unix socket
+            UnixSocketAddress unixAddress = new UnixSocketAddress( this._unixSocket );
+            channel = UnixSocketChannel.open( unixAddress );
+
+            HttpDestination destination = (HttpDestination)context.get(HTTP_DESTINATION_CONTEXT_KEY);
+            HttpClient client = destination.getHttpClient();
+
+            configure(client, channel);
+
+            channel.configureBlocking(false);
+            selectorManager.accept(channel, context);            
+        }
+        // Must catch all exceptions, since some like
+        // UnresolvedAddressException are not IOExceptions.
+        catch (Throwable x)
+        {
+            // If IPv6 is not deployed, a generic SocketException "Network is unreachable"
+            // exception is being thrown, so we attempt to provide a better error message.
+            if (x.getClass() == SocketException.class)
+                x = new SocketException("Could not connect to " + address).initCause(x);
+
+            try
+            {
+                if (channel != null)
+                    channel.close();
+            }
+            catch (IOException xx)
+            {
+                LOG.ignore(xx);
+            }
+            finally
+            {
+                connectFailed(context, x);
+            }
+        }
+    }
+
+    public class UnixSocketSelectorManager extends ClientSelectorManager
+    {
+        protected UnixSocketSelectorManager(HttpClient client, int selectors)
+        {
+            super(client,selectors);
+        }
+
+
+        @Override
+        protected Selector newSelector() throws IOException
+        {
+            return NativeSelectorProvider.getInstance().openSelector();
+        }
+
+        @Override
+        protected EndPoint newEndPoint(SelectableChannel channel, ManagedSelector selector, SelectionKey key)
+        {
+            UnixSocketEndPoint endp = new UnixSocketEndPoint((UnixSocketChannel)channel, selector, key, getScheduler());
+            endp.setIdleTimeout(getHttpClient().getIdleTimeout());
+            return endp;
+        }
+
+
+
+    }
+}