Skip to content

Commit 97703d9

Browse files
rschmittrschmitt
committed
Enable JEP 380 UDS support for sync client
This change enables the sync client to use standard library Unix-domain sockets on Java 16+ without a JUnixSocket dependency, as is already supported by the async client. JEP 380 only exposes Unix-domain socket support through the `SocketChannel` API, but the sync client is tightly coupled to the legacy `java.net.Socket` API. The idea behind this change is to wrap the UDS `SocketChannel` in a subclass of `java.net.Socket` that acts as a translation layer between the legacy socket API and `SocketChannel`. In particular, the adapter adds support for socket timeouts by translating blocking reads to non-blocking reads and supplying the socket timeout value to the `select()` call. After this change, HttpClient will use the Java standard library by default. JUnixSocket will only be loaded on older versions of Java that lack standard library support for UDS, and even then only for the synchronous client, as the JUnixSocket-provided `SocketChannel` cannot be used with the JDK-provided `Selector` used by `IOReactor`.
1 parent 239948e commit 97703d9

3 files changed

Lines changed: 347 additions & 11 deletions

File tree

‎httpclient5/src/main/java/org/apache/hc/client5/http/socket/Jep380SocketChannelAdapter.java‎

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,95 @@
1+
/*
2+
* ====================================================================
3+
* Licensed to the Apache Software Foundation (ASF) under one
4+
* or more contributor license agreements. See the NOTICE file
5+
* distributed with this work for additional information
6+
* regarding copyright ownership. The ASF licenses this file
7+
* to you under the Apache License, Version 2.0 (the
8+
* "License"); you may not use this file except in compliance
9+
* with the License. You may obtain a copy of the License at
10+
*
11+
* http://www.apache.org/licenses/LICENSE-2.0
12+
*
13+
* Unless required by applicable law or agreed to in writing,
14+
* software distributed under the License is distributed on an
15+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
16+
* KIND, either express or implied. See the License for the
17+
* specific language governing permissions and limitations
18+
* under the License.
19+
* ====================================================================
20+
*
21+
* This software consists of voluntary contributions made by many
22+
* individuals on behalf of the Apache Software Foundation. For more
23+
* information on the Apache Software Foundation, please see
24+
* <http://www.apache.org/>.
25+
*
26+
*/
27+
28+
package org.apache.hc.client5.http.socket;
29+
30+
import java.io.IOException;
31+
import java.io.InputStream;
32+
import java.io.OutputStream;
33+
import java.net.Socket;
34+
import java.net.SocketAddress;
35+
import java.net.SocketException;
36+
import java.nio.channels.SocketChannel;
37+
38+
final class Jep380SocketChannelAdapter extends Socket {
39+
private final SocketChannel channel;
40+
private final Jep380SocketChannelImplAdapter adapter;
41+
42+
Jep380SocketChannelAdapter(final SocketChannel channel) throws IOException {
43+
this(channel, new Jep380SocketChannelImplAdapter(channel));
44+
}
45+
46+
private Jep380SocketChannelAdapter(final SocketChannel channel, final Jep380SocketChannelImplAdapter adapter) {
47+
this.channel = channel;
48+
this.adapter = adapter;
49+
}
50+
51+
@Override
52+
public void connect(final SocketAddress endpoint, final int timeout) throws IOException {
53+
channel.connect(endpoint);
54+
}
55+
56+
@Override
57+
public void connect(final SocketAddress endpoint) throws IOException {
58+
channel.connect(endpoint);
59+
}
60+
61+
@Override
62+
public boolean isConnected() {
63+
return channel.isConnected();
64+
}
65+
66+
@Override
67+
public InputStream getInputStream() throws IOException {
68+
return adapter.getInputStream();
69+
}
70+
71+
@Override
72+
public OutputStream getOutputStream() throws IOException {
73+
return adapter.getOutputStream();
74+
}
75+
76+
@Override
77+
public boolean isClosed() {
78+
return !channel.isOpen();
79+
}
80+
81+
@Override
82+
public void close() throws IOException {
83+
adapter.close();
84+
}
85+
86+
@Override
87+
public int getSoTimeout() throws SocketException {
88+
return adapter.soTimeoutMs;
89+
}
90+
91+
@Override
92+
public void setSoTimeout(final int timeout) throws SocketException {
93+
adapter.soTimeoutMs = timeout;
94+
}
95+
}

‎httpclient5/src/main/java/org/apache/hc/client5/http/socket/Jep380SocketChannelImplAdapter.java‎

Lines changed: 232 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,232 @@
1+
/*
2+
* ====================================================================
3+
* Licensed to the Apache Software Foundation (ASF) under one
4+
* or more contributor license agreements. See the NOTICE file
5+
* distributed with this work for additional information
6+
* regarding copyright ownership. The ASF licenses this file
7+
* to you under the Apache License, Version 2.0 (the
8+
* "License"); you may not use this file except in compliance
9+
* with the License. You may obtain a copy of the License at
10+
*
11+
* http://www.apache.org/licenses/LICENSE-2.0
12+
*
13+
* Unless required by applicable law or agreed to in writing,
14+
* software distributed under the License is distributed on an
15+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
16+
* KIND, either express or implied. See the License for the
17+
* specific language governing permissions and limitations
18+
* under the License.
19+
* ====================================================================
20+
*
21+
* This software consists of voluntary contributions made by many
22+
* individuals on behalf of the Apache Software Foundation. For more
23+
* information on the Apache Software Foundation, please see
24+
* <http://www.apache.org/>.
25+
*
26+
*/
27+
28+
package org.apache.hc.client5.http.socket;
29+
30+
import java.io.IOException;
31+
import java.io.InputStream;
32+
import java.io.OutputStream;
33+
import java.io.UncheckedIOException;
34+
import java.net.InetAddress;
35+
import java.net.SocketAddress;
36+
import java.net.SocketException;
37+
import java.net.SocketImpl;
38+
import java.net.SocketOptions;
39+
import java.net.SocketTimeoutException;
40+
import java.net.StandardSocketOptions;
41+
import java.nio.ByteBuffer;
42+
import java.nio.channels.SelectionKey;
43+
import java.nio.channels.Selector;
44+
import java.nio.channels.SocketChannel;
45+
46+
final class Jep380SocketChannelImplAdapter extends SocketImpl {
47+
private final SocketChannel channel;
48+
volatile int soTimeoutMs = 0;
49+
50+
public Jep380SocketChannelImplAdapter(final SocketChannel channel) throws IOException {
51+
this.channel = channel;
52+
channel.configureBlocking(false);
53+
}
54+
55+
@Override
56+
protected void close() throws IOException {
57+
channel.close();
58+
}
59+
60+
@Override
61+
protected InputStream getInputStream() throws IOException {
62+
return new InputStreamAdapter();
63+
}
64+
65+
@Override
66+
protected OutputStream getOutputStream() throws IOException {
67+
return new OutputStreamAdapter();
68+
}
69+
70+
@Override
71+
public Object getOption(final int optID) throws SocketException {
72+
try {
73+
switch (optID) {
74+
case SocketOptions.SO_TIMEOUT:
75+
return soTimeoutMs;
76+
case SocketOptions.SO_RCVBUF:
77+
return channel.getOption(StandardSocketOptions.SO_RCVBUF);
78+
case SocketOptions.SO_SNDBUF:
79+
return channel.getOption(StandardSocketOptions.SO_SNDBUF);
80+
}
81+
} catch (final IOException ex) {
82+
throw new UncheckedIOException(ex);
83+
}
84+
throw new UnsupportedOperationException("getOption: " + optID);
85+
}
86+
87+
@Override
88+
public void setOption(final int optID, final Object value) throws SocketException {
89+
try {
90+
switch (optID) {
91+
case SocketOptions.SO_TIMEOUT:
92+
soTimeoutMs = (Integer) value;
93+
return;
94+
case SocketOptions.SO_RCVBUF:
95+
channel.setOption(StandardSocketOptions.SO_RCVBUF, (Integer) value);
96+
return;
97+
case SocketOptions.SO_SNDBUF:
98+
channel.setOption(StandardSocketOptions.SO_SNDBUF, (Integer) value);
99+
return;
100+
}
101+
} catch (final IOException ex) {
102+
throw new RuntimeException(ex);
103+
}
104+
throw new UnsupportedOperationException();
105+
}
106+
107+
@Override
108+
protected void accept(final SocketImpl s) throws IOException {
109+
throw new UnsupportedOperationException();
110+
}
111+
112+
@Override
113+
protected int available() throws IOException {
114+
throw new UnsupportedOperationException();
115+
}
116+
117+
@Override
118+
protected void bind(final InetAddress host, final int port) throws IOException {
119+
throw new UnsupportedOperationException();
120+
}
121+
122+
@Override
123+
protected void connect(final String host, final int port) throws IOException {
124+
throw new UnsupportedOperationException();
125+
}
126+
127+
@Override
128+
protected void connect(final InetAddress address, final int port) throws IOException {
129+
throw new UnsupportedOperationException();
130+
}
131+
132+
@Override
133+
protected void connect(final SocketAddress address, final int timeout) throws IOException {
134+
throw new UnsupportedOperationException();
135+
}
136+
137+
@Override
138+
protected void create(final boolean stream) throws IOException {
139+
throw new UnsupportedOperationException();
140+
}
141+
142+
@Override
143+
protected void listen(final int backlog) throws IOException {
144+
throw new UnsupportedOperationException();
145+
}
146+
147+
@Override
148+
protected void sendUrgentData(final int data) throws IOException {
149+
throw new UnsupportedOperationException();
150+
}
151+
152+
private class InputStreamAdapter extends InputStream {
153+
private final Selector sel;
154+
private final SelectionKey key;
155+
156+
private InputStreamAdapter() throws IOException {
157+
this.sel = Selector.open();
158+
this.key = channel.register(sel, SelectionKey.OP_READ);
159+
}
160+
161+
@Override
162+
public int read() throws IOException {
163+
final byte[] b = new byte[1];
164+
final int n = read(b, 0, 1);
165+
return (n == -1) ? -1 : (b[0] & 0xFF);
166+
}
167+
168+
@Override
169+
public int read(final byte[] b) throws IOException {
170+
return read(b, 0, b.length);
171+
}
172+
173+
@Override
174+
public int read(final byte[] b, final int off, final int len) throws IOException {
175+
final ByteBuffer buf = ByteBuffer.wrap(b, off, len);
176+
if (sel.select(soTimeoutMs) == 0) {
177+
throw new SocketTimeoutException();
178+
}
179+
final int read = channel.read(buf);
180+
sel.selectedKeys().clear();
181+
return read;
182+
}
183+
184+
@Override
185+
public void close() throws IOException {
186+
key.cancel();
187+
sel.close();
188+
channel.close();
189+
}
190+
}
191+
192+
private class OutputStreamAdapter extends OutputStream {
193+
private final Selector sel;
194+
private final SelectionKey key;
195+
196+
private OutputStreamAdapter() throws IOException {
197+
this.sel = Selector.open();
198+
this.key = channel.register(sel, SelectionKey.OP_WRITE);
199+
}
200+
201+
@Override
202+
public void write(final int b) throws IOException {
203+
write(new byte[]{ (byte) b});
204+
}
205+
206+
@Override
207+
public void write(final byte[] b) throws IOException {
208+
write(b, 0, b.length);
209+
}
210+
211+
@Override
212+
public void write(final byte[] b, final int off, final int len) throws IOException {
213+
final ByteBuffer buf = ByteBuffer.wrap(b, off, len);
214+
while (buf.hasRemaining()) {
215+
final int n = channel.write(buf);
216+
if (n == 0) {
217+
if (sel.select(60_000) == 0) {
218+
throw new SocketTimeoutException("write timed out");
219+
}
220+
sel.selectedKeys().clear();
221+
}
222+
}
223+
}
224+
225+
@Override
226+
public void close() throws IOException {
227+
key.cancel();
228+
sel.close();
229+
channel.close();
230+
}
231+
}
232+
}

‎httpclient5/src/main/java/org/apache/hc/client5/http/socket/UnixDomainSocketFactory.java‎

Lines changed: 20 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -37,8 +37,11 @@
3737

3838
import java.io.IOException;
3939
import java.lang.reflect.Method;
40+
import java.net.ProtocolFamily;
4041
import java.net.Socket;
4142
import java.net.SocketAddress;
43+
import java.net.StandardProtocolFamily;
44+
import java.nio.channels.SocketChannel;
4245
import java.nio.file.Path;
4346

4447
/**
@@ -70,14 +73,14 @@ private enum Implementation {
7073

7174
private static Implementation detectImplementation() {
7275
try {
73-
Class.forName(JUNIXSOCKET_SOCKET_CLASS);
74-
LOG.debug("Using JUnixSocket Unix Domain Socket implementation");
75-
return Implementation.JUNIXSOCKET;
76+
Class.forName(JDK_UNIX_SOCKET_ADDRESS_CLASS);
77+
LOG.debug("Using JDK Unix Domain Socket implementation");
78+
return Implementation.JDK;
7679
} catch (final ClassNotFoundException e) {
7780
try {
78-
Class.forName(JDK_UNIX_SOCKET_ADDRESS_CLASS);
79-
LOG.debug("Using JDK Unix Domain Socket implementation");
80-
return Implementation.JDK;
81+
Class.forName(JUNIXSOCKET_SOCKET_CLASS);
82+
LOG.debug("Using JUnixSocket Unix Domain Socket implementation");
83+
return Implementation.JUNIXSOCKET;
8184
} catch (final ClassNotFoundException e2) {
8285
LOG.debug("No Unix Domain Socket implementation found");
8386
return Implementation.NONE;
@@ -138,11 +141,17 @@ public Socket createSocket() throws IOException {
138141

139142
try {
140143
if (IMPLEMENTATION == Implementation.JDK) {
141-
// Java 16+ only supports UDS through the SocketChannel API, but the sync client is coupled
142-
// to the legacy Socket API. In order to use Java sockets, we first need to write an
143-
// adapter, similar to the one provided by JUnixSocket.
144-
throw new UnsupportedOperationException("JEP 380 Unix domain sockets are not supported; use "
145-
+ "JUnixSocket");
144+
// Java 16+ only supports UDS through the SocketChannel API, but the sync client is coupled to the
145+
// legacy Socket API. To facilitate this, we use an adapter, similar to the one provided by JUnixSocket.
146+
try {
147+
final SocketChannel channel = (SocketChannel) SocketChannel.class.getMethod("open",
148+
ProtocolFamily.class)
149+
.invoke(null, StandardProtocolFamily.valueOf("UNIX"));
150+
return new Jep380SocketChannelAdapter(channel);
151+
} catch (final ReflectiveOperationException ex) {
152+
throw new UnsupportedOperationException("JEP 380 Unix domain sockets are not supported; use "
153+
+ "JUnixSocket", ex);
154+
}
146155
} else {
147156
// JUnixSocket implementation
148157
final Class<?> socketClass = Class.forName(JUNIXSOCKET_SOCKET_CLASS);

0 commit comments

Comments
 (0)