h4ckx0r/engine.io-client-java
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Options for inject okhttp Call/WebSocket factory

Browse Source
This commit is contained in:
Yu-Hsuan Lin
2017-05-02 01:15:47 +08:00
parent 80535c6304
commit c817af5b22
9 changed files with 183 additions and 267 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
src/main/java/io/socket/engineio/client/Socket.java
View File

@@ -1,5 +1,21 @@
package io.socket.engineio.client;
import org.json.JSONException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;
import io.socket.emitter.Emitter;
import io.socket.engineio.client.transports.Polling;
import io.socket.engineio.client.transports.PollingXHR;
@@ -8,19 +24,7 @@ import io.socket.engineio.parser.Packet;
import io.socket.engineio.parser.Parser;
import io.socket.parseqs.ParseQS;
import io.socket.thread.EventThread;
import org.json.JSONException;
import javax.net.ssl.HostnameVerifier;
import javax.net.ssl.SSLContext;
import java.net.Proxy;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;
import okhttp3.OkHttpClient;
/**
@@ -98,8 +102,9 @@ public class Socket extends Emitter {
private static boolean priorWebsocketSuccess = false;
private static SSLContext defaultSSLContext;
private static HostnameVerifier defaultHostnameVerifier;
private static okhttp3.WebSocket.Factory defaultWebSocketFactory;
private static okhttp3.Call.Factory defaultCallFactory;
private static OkHttpClient defaultOkHttpClient;
private boolean secure;
private boolean upgrade;
@@ -122,11 +127,8 @@ public class Socket extends Emitter {
/*package*/ Transport transport;
private Future pingTimeoutTimer;
private Future pingIntervalTimer;
private SSLContext sslContext;
private HostnameVerifier hostnameVerifier;
public Proxy proxy;
public String proxyLogin;
public String proxyPassword;
private okhttp3.WebSocket.Factory webSocketFactory;
private okhttp3.Call.Factory callFactory;
private ReadyState readyState;
private ScheduledExecutorService heartbeatScheduler;
@@ -190,7 +192,6 @@ public class Socket extends Emitter {
opts.port = this.secure ? 443 : 80;
}
this.sslContext = opts.sslContext != null ? opts.sslContext : defaultSSLContext;
this.hostname = opts.hostname != null ? opts.hostname : "localhost";
this.port = opts.port;
this.query = opts.query != null ?
@@ -203,18 +204,28 @@ public class Socket extends Emitter {
opts.transports : new String[]{Polling.NAME, WebSocket.NAME}));
this.policyPort = opts.policyPort != 0 ? opts.policyPort : 843;
this.rememberUpgrade = opts.rememberUpgrade;
this.hostnameVerifier = opts.hostnameVerifier != null ? opts.hostnameVerifier : defaultHostnameVerifier;
this.proxy = opts.proxy;
this.proxyLogin = opts.proxyLogin;
this.proxyPassword = opts.proxyPassword;
this.callFactory = opts.callFactory != null ? opts.callFactory : defaultCallFactory;
this.webSocketFactory = opts.webSocketFactory != null ? opts.webSocketFactory : defaultWebSocketFactory;
if (callFactory == null) {
if (defaultOkHttpClient == null) {
defaultOkHttpClient = new OkHttpClient();
}
callFactory = defaultOkHttpClient;
}
if (webSocketFactory == null) {
if (defaultOkHttpClient == null) {
defaultOkHttpClient = new OkHttpClient();
}
webSocketFactory = defaultOkHttpClient;
}
}
public static void setDefaultSSLContext(SSLContext sslContext) {
defaultSSLContext = sslContext;
public static void setDefaultOkHttpWebSocketFactory(okhttp3.WebSocket.Factory factory) {
defaultWebSocketFactory = factory;
}
public static void setDefaultHostnameVerifier(HostnameVerifier hostnameVerifier) {
defaultHostnameVerifier = hostnameVerifier;
public static void setDefaultOkHttpCallFactory(okhttp3.Call.Factory factory) {
defaultCallFactory = factory;
}
/**
@@ -262,7 +273,6 @@ public class Socket extends Emitter {
}
Transport.Options opts = new Transport.Options();
opts.sslContext = this.sslContext;
opts.hostname = this.hostname;
opts.port = this.port;
opts.secure = this.secure;
@@ -272,10 +282,8 @@ public class Socket extends Emitter {
opts.timestampParam = this.timestampParam;
opts.policyPort = this.policyPort;
opts.socket = this;
opts.hostnameVerifier = this.hostnameVerifier;
opts.proxy = this.proxy;
opts.proxyLogin = this.proxyLogin;
opts.proxyPassword = this.proxyPassword;
opts.callFactory = this.callFactory;
opts.webSocketFactory = this.webSocketFactory;
Transport transport;
if (WebSocket.NAME.equals(name)) {

31
src/main/java/io/socket/engineio/client/Transport.java
View File

@@ -1,16 +1,15 @@
package io.socket.engineio.client;
import java.util.Map;
import io.socket.emitter.Emitter;
import io.socket.engineio.parser.Packet;
import io.socket.engineio.parser.Parser;
import io.socket.thread.EventThread;
import io.socket.utf8.UTF8Exception;
import javax.net.ssl.HostnameVerifier;
import javax.net.ssl.SSLContext;
import java.net.Proxy;
import java.util.Map;
import okhttp3.Call;
import okhttp3.WebSocket;
public abstract class Transport extends Emitter {
@@ -41,14 +40,10 @@ public abstract class Transport extends Emitter {
protected String path;
protected String hostname;
protected String timestampParam;
protected SSLContext sslContext;
protected Socket socket;
protected HostnameVerifier hostnameVerifier;
protected Proxy proxy;
protected String proxyLogin;
protected String proxyPassword;
protected ReadyState readyState;
protected WebSocket.Factory webSocketFactory;
protected Call.Factory callFactory;
public Transport(Options opts) {
this.path = opts.path;
@@ -58,12 +53,9 @@ public abstract class Transport extends Emitter {
this.query = opts.query;
this.timestampParam = opts.timestampParam;
this.timestampRequests = opts.timestampRequests;
this.sslContext = opts.sslContext;
this.socket = opts.socket;
this.hostnameVerifier = opts.hostnameVerifier;
this.proxy = opts.proxy;
this.proxyLogin = opts.proxyLogin;
this.proxyPassword = opts.proxyPassword;
this.webSocketFactory = opts.webSocketFactory;
this.callFactory = opts.callFactory;
}
protected Transport onError(String msg, Exception desc) {
@@ -156,11 +148,8 @@ public abstract class Transport extends Emitter {
public int port = -1;
public int policyPort = -1;
public Map<String, String> query;
public SSLContext sslContext;
public HostnameVerifier hostnameVerifier;
protected Socket socket;
public Proxy proxy;
public String proxyLogin;
public String proxyPassword;
public WebSocket.Factory webSocketFactory;
public Call.Factory callFactory;
}
}

176
src/main/java/io/socket/engineio/client/transports/PollingXHR.java
View File

@@ -1,20 +1,25 @@
package io.socket.engineio.client.transports;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.logging.Logger;
import io.socket.emitter.Emitter;
import io.socket.engineio.client.Transport;
import io.socket.thread.EventThread;
import javax.net.ssl.HostnameVerifier;
import javax.net.ssl.HttpsURLConnection;
import javax.net.ssl.SSLContext;
import java.io.*;
import java.net.HttpURLConnection;
import java.net.Proxy;
import java.net.URL;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.logging.Logger;
import okhttp3.Call;
import okhttp3.Callback;
import okhttp3.HttpUrl;
import okhttp3.MediaType;
import okhttp3.OkHttpClient;
import okhttp3.RequestBody;
import okhttp3.Response;
public class PollingXHR extends Polling {
@@ -33,9 +38,7 @@ public class PollingXHR extends Polling {
opts = new Request.Options();
}
opts.uri = this.uri();
opts.sslContext = this.sslContext;
opts.hostnameVerifier = this.hostnameVerifier;
opts.proxy = this.proxy;
opts.callFactory = this.callFactory;
Request req = new Request(opts);
@@ -136,6 +139,7 @@ public class PollingXHR extends Polling {
public static final String EVENT_ERROR = "error";
public static final String EVENT_REQUEST_HEADERS = "requestHeaders";
public static final String EVENT_RESPONSE_HEADERS = "responseHeaders";
private static final String BINARY_CONTENT_TYPE = "application/octet-stream";
private String method;
private String uri;
@@ -143,93 +147,60 @@ public class PollingXHR extends Polling {
// data is always a binary
private byte[] data;
private SSLContext sslContext;
private HttpURLConnection xhr;
private HostnameVerifier hostnameVerifier;
private Proxy proxy;
private Call.Factory callFactory;
private Response response;
private Call requestCall;
public Request(Options opts) {
this.method = opts.method != null ? opts.method : "GET";
this.uri = opts.uri;
this.data = opts.data;
this.sslContext = opts.sslContext;
this.hostnameVerifier = opts.hostnameVerifier;
this.proxy = opts.proxy;
this.callFactory = opts.callFactory != null ? opts.callFactory : new OkHttpClient();
}
public void create() {
final Request self = this;
try {
logger.fine(String.format("xhr open %s: %s", this.method, this.uri));
URL url = new URL(this.uri);
xhr = proxy != null ? (HttpURLConnection) url.openConnection(proxy)
: (HttpURLConnection) url.openConnection();
xhr.setRequestMethod(this.method);
} catch (IOException e) {
this.onError(e);
return;
}
xhr.setConnectTimeout(10000);
if (xhr instanceof HttpsURLConnection) {
if (this.sslContext != null) {
((HttpsURLConnection)xhr).setSSLSocketFactory(this.sslContext.getSocketFactory());
}
if (this.hostnameVerifier != null) {
((HttpsURLConnection)xhr).setHostnameVerifier(this.hostnameVerifier);
}
}
logger.fine(String.format("xhr open %s: %s", this.method, this.uri));
Map<String, List<String>> headers = new TreeMap<String, List<String>>(String.CASE_INSENSITIVE_ORDER);
if ("POST".equals(this.method)) {
xhr.setDoOutput(true);
headers.put("Content-type", new LinkedList<String>(Arrays.asList("application/octet-stream")));
headers.put("Content-type", new LinkedList<String>(Collections.singletonList(BINARY_CONTENT_TYPE)));
}
self.onRequestHeaders(headers);
logger.fine(String.format("sending xhr with url %s | data %s", this.uri, Arrays.toString(this.data)));
okhttp3.Request.Builder requestBuilder = new okhttp3.Request.Builder();
for (Map.Entry<String, List<String>> header : headers.entrySet()) {
for (String v : header.getValue()){
xhr.addRequestProperty(header.getKey(), v);
requestBuilder.addHeader(header.getKey(), v);
}
}
okhttp3.Request request = requestBuilder
.url(HttpUrl.parse(self.uri))
.method(self.method, (self.data != null) ?
RequestBody.create(MediaType.parse(BINARY_CONTENT_TYPE), self.data) : null)
.build();
logger.fine(String.format("sending xhr with url %s | data %s", this.uri, this.data));
new Thread(new Runnable() {
requestCall = callFactory.newCall(request);
requestCall.enqueue(new Callback() {
@Override
public void run() {
OutputStream output = null;
try {
if (self.data != null) {
xhr.setFixedLengthStreamingMode(self.data.length);
output = new BufferedOutputStream(xhr.getOutputStream());
output.write(self.data);
output.flush();
}
public void onFailure(Call call, IOException e) {
self.onError(e);
}
Map<String, List<String>> headers = xhr.getHeaderFields();
self.onResponseHeaders(headers);
@Override
public void onResponse(Call call, Response response) throws IOException {
self.response = response;
self.onResponseHeaders(response.headers().toMultimap());
final int statusCode = xhr.getResponseCode();
if (HttpURLConnection.HTTP_OK == statusCode) {
self.onLoad();
} else {
self.onError(new IOException(Integer.toString(statusCode)));
}
} catch (IOException e) {
self.onError(e);
} catch (NullPointerException e) {
// It would occur to disconnect
// https://code.google.com/p/android/issues/detail?id=76592
self.onError(e);
} finally {
try {
if (output != null) output.close();
} catch (IOException e) {}
if (response.isSuccessful()) {
self.onLoad();
} else {
self.onError(new IOException(Integer.toString(response.code())));
}
}
}).start();
});
}
private void onSuccess() {
@@ -258,56 +229,17 @@ public class PollingXHR extends Polling {
this.emit(EVENT_RESPONSE_HEADERS, headers);
}
private void cleanup() {
if (xhr == null) {
return;
}
xhr.disconnect();
xhr = null;
}
private void onLoad() {
InputStream input = null;
BufferedReader reader = null;
String contentType = xhr.getContentType();
String contentType = response.body().contentType().toString();
try {
if ("application/octet-stream".equalsIgnoreCase(contentType)) {
input = new BufferedInputStream(this.xhr.getInputStream());
List<byte[]> buffers = new ArrayList<byte[]>();
int capacity = 0;
int len = 0;
byte[] buffer = new byte[1024];
while ((len = input.read(buffer)) > 0) {
byte[] tempBuffer = new byte[len];
System.arraycopy(buffer, 0, tempBuffer, 0, len);
buffers.add(tempBuffer);
capacity += len;
}
ByteBuffer data = ByteBuffer.allocate(capacity);
for (byte[] b : buffers) {
data.put(b);
}
this.onData(data.array());
if (BINARY_CONTENT_TYPE.equalsIgnoreCase(contentType)) {
this.onData(response.body().bytes());
} else {
String line;
StringBuilder data = new StringBuilder();
reader = new BufferedReader(new InputStreamReader(xhr.getInputStream()));
while ((line = reader.readLine()) != null) {
data.append(line);
}
this.onData(data.toString());
this.onData(response.body().string());
}
} catch (IOException e) {
this.onError(e);
} finally {
try {
if (input != null) input.close();
} catch (IOException e) {}
try {
if (reader != null) reader.close();
} catch (IOException e) {}
this.cleanup();
}
}
@@ -316,9 +248,7 @@ public class PollingXHR extends Polling {
public String uri;
public String method;
public byte[] data;
public SSLContext sslContext;
public HostnameVerifier hostnameVerifier;
public Proxy proxy;
public Call.Factory callFactory;
}
}
}

63
src/main/java/io/socket/engineio/client/transports/WebSocket.java
View File

@@ -1,6 +1,12 @@
package io.socket.engineio.client.transports;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.logging.Logger;
import io.socket.engineio.client.Transport;
import io.socket.engineio.parser.Packet;
import io.socket.engineio.parser.Parser;
@@ -8,18 +14,12 @@ import io.socket.parseqs.ParseQS;
import io.socket.thread.EventThread;
import io.socket.utf8.UTF8Exception;
import io.socket.yeast.Yeast;
import okhttp3.*;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.WebSocketListener;
import okio.ByteString;
import javax.net.ssl.SSLSocketFactory;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;
public class WebSocket extends Transport {
@@ -39,34 +39,7 @@ public class WebSocket extends Transport {
this.emit(EVENT_REQUEST_HEADERS, headers);
final WebSocket self = this;
OkHttpClient.Builder clientBuilder = new OkHttpClient.Builder()
// turn off timeouts (github.com/socketio/engine.io-client-java/issues/32)
.connectTimeout(0, TimeUnit.MILLISECONDS)
.readTimeout(0, TimeUnit.MILLISECONDS)
.writeTimeout(0, TimeUnit.MILLISECONDS);
if (this.sslContext != null) {
SSLSocketFactory factory = sslContext.getSocketFactory();// (SSLSocketFactory) SSLSocketFactory.getDefault();
clientBuilder.sslSocketFactory(factory);
}
if (this.hostnameVerifier != null) {
clientBuilder.hostnameVerifier(this.hostnameVerifier);
}
if (proxy != null) {
clientBuilder.proxy(proxy);
}
if (proxyLogin != null && !proxyLogin.isEmpty()) {
final String credentials = Credentials.basic(proxyLogin, proxyPassword);
clientBuilder.proxyAuthenticator(new Authenticator() {
@Override
public Request authenticate(Route route, Response response) throws IOException {
return response.request().newBuilder()
.header("Proxy-Authorization", credentials)
.build();
}
});
}
okhttp3.WebSocket.Factory factory = webSocketFactory != null ? webSocketFactory : new OkHttpClient();
Request.Builder builder = new Request.Builder().url(uri());
for (Map.Entry<String, List<String>> entry : headers.entrySet()) {
for (String v : entry.getValue()) {
@@ -74,8 +47,7 @@ public class WebSocket extends Transport {
}
}
final Request request = builder.build();
final OkHttpClient client = clientBuilder.build();
ws = client.newWebSocket(request, new WebSocketListener() {
ws = factory.newWebSocket(request, new WebSocketListener() {
@Override
public void onOpen(okhttp3.WebSocket webSocket, Response response) {
final Map<String, List<String>> headers = response.headers().toMultimap();
@@ -137,7 +109,6 @@ public class WebSocket extends Transport {
});
}
});
client.dispatcher().executorService().shutdown();
}
protected void write(Packet[] packets) throws UTF8Exception {
@@ -187,14 +158,8 @@ public class WebSocket extends Transport {
protected void doClose() {
if (ws != null) {
try {
ws.close(1000, "");
} catch (IllegalStateException e) {
// websocket already closed
}
}
if (ws != null) {
ws.cancel();
ws.close(1000, "");
ws = null;
}
}