Skip to content

A
AloqaIM-Android
Commit abe34222 authored by Leonardo Aramaki's avatar Leonardo Aramaki
Browse files
Options

Fix reconnection loop issue when add new server

parent 65d5c3a7
Hide whitespace changes
Inline Side-by-side
Showing with 527 additions and 459 deletions
android-ddp/src/main/java/chat/rocket/android_ddp/DDPClientImpl.java
View file @ abe34222
......@@ -22,440 +22,467 @@ import io.reactivex.disposables.CompositeDisposable;
import okhttp3.OkHttpClient;
public class DDPClientImpl {
private final DDPClient client;
private RxWebSocket websocket;
private Flowable<RxWebSocketCallback.Base> flowable;
private CompositeDisposable disposables;
private String currentSession;
/* package */ DDPClientImpl(DDPClient self, OkHttpClient client) {
websocket = new RxWebSocket(client);
this.client = self;
}
private static JSONObject toJson(String s) {
if (TextUtils.isEmpty(s)) {
return null;
private final DDPClient client;
private RxWebSocket websocket;
private Flowable<RxWebSocketCallback.Base> flowable;
private CompositeDisposable disposables;
private String currentSession;
/* package */ DDPClientImpl(DDPClient self, OkHttpClient client) {
websocket = new RxWebSocket(client);
this.client = self;
}
try {
return new JSONObject(s);
} catch (JSONException e) {
return null;
}
}
private static String extractMsg(JSONObject response) {
if (response == null || response.isNull("msg")) {
return null;
} else {
return response.optString("msg");
private static JSONObject toJson(String s) {
if (TextUtils.isEmpty(s)) {
return null;
}
try {
return new JSONObject(s);
} catch (JSONException e) {
return null;
}
}
}
/* package */ void connect(final TaskCompletionSource<DDPClientCallback.Connect> task, final String url,
String session) {
try {
flowable = websocket.connect(url).autoConnect(2);
CompositeDisposable disposables = new CompositeDisposable();
private static String extractMsg(JSONObject response) {
if (response == null || response.isNull("msg")) {
return null;
} else {
return response.optString("msg");
}
}
disposables.add(
flowable.filter(callback -> callback instanceof RxWebSocketCallback.Open)
.subscribe(
callback ->
sendMessage("connect",
json -> (TextUtils.isEmpty(session) ? json : json.put("session", DDPClientImpl.this.currentSession))
.put(
"version", "1")
.put("support", new JSONArray().put("1").put("pre2").put("pre1")),
task),
RCLog::e
)
);
disposables.add(
flowable.filter(
callback -> callback instanceof RxWebSocketCallback.Message)
.map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
.map(DDPClientImpl::toJson)
.timeout(7, TimeUnit.SECONDS)
.subscribe(response -> {
String msg = extractMsg(response);
if ("connected".equals(msg) && !response.isNull("session")) {
currentSession = response.optString("session");
task.trySetResult(
new DDPClientCallback.Connect(client, response.optString("session")));
disposables.clear();
} else if ("error".equals(msg) && "Already connected".equals(
response.optString("reason"))) {
task.trySetResult(new DDPClientCallback.Connect(client, null));
disposables.clear();
} else if ("failed".equals(msg)) {
task.trySetError(
new DDPClientCallback.Connect.Failed(client, response.optString("version")));
disposables.clear();
}
},
err -> task.trySetError(new DDPClientCallback.Connect.Timeout(client))
)
);
/* package */ void connect(final TaskCompletionSource<DDPClientCallback.Connect> task, final String url,
String session) {
try {
flowable = websocket.connect(url).autoConnect(2);
CompositeDisposable disposables = new CompositeDisposable();
disposables.add(
flowable.filter(callback -> callback instanceof RxWebSocketCallback.Open)
.subscribe(
callback ->
sendMessage("connect",
json -> (TextUtils.isEmpty(session) ? json : json.put("session", DDPClientImpl.this.currentSession))
.put(
"version", "1")
.put("support", new JSONArray().put("1").put("pre2").put("pre1")),
task),
RCLog::e
)
);
disposables.add(
flowable.filter(
callback -> callback instanceof RxWebSocketCallback.Message)
.map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
.map(DDPClientImpl::toJson)
.timeout(7, TimeUnit.SECONDS)
.subscribe(response -> {
String msg = extractMsg(response);
if ("connected".equals(msg) && !response.isNull("session")) {
currentSession = response.optString("session");
task.trySetResult(
new DDPClientCallback.Connect(client, response.optString("session")));
disposables.clear();
} else if ("error".equals(msg) && "Already connected".equals(
response.optString("reason"))) {
task.trySetResult(new DDPClientCallback.Connect(client, null));
disposables.clear();
} else if ("failed".equals(msg)) {
task.trySetError(
new DDPClientCallback.Connect.Failed(client, response.optString("version")));
disposables.clear();
}
},
err -> {
if (err instanceof TimeoutException) {
task.trySetError(new Exception("Your connection seems off…"));
} else {
task.trySetError(new Exception("Ooops. Something's up!"));
}
}
)
);
addErrorCallback(disposables, task);
addErrorCallback(disposables, task);
subscribeBaseListeners();
} catch (Exception e) {
RCLog.e(e);
subscribeBaseListeners();
} catch (Exception e) {
RCLog.e(e);
}
}
}
/* package */ Maybe<DDPClientCallback.Base> ping(@Nullable final String id) {
final boolean requested = (TextUtils.isEmpty(id)) ?
sendMessage("ping", null) :
sendMessage("ping", json -> json.put("id", id));
if (requested) {
return flowable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
.timeout(8, TimeUnit.SECONDS)
.map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
.map(DDPClientImpl::toJson)
.filter(response -> "pong".equalsIgnoreCase(extractMsg(response)))
.doOnError(error -> {
RCLog.e(error, "Heartbeat ping[%s] xxx failed xxx", id);
})
.map(response -> {
String msg = extractMsg(response);
if ("pong".equals(msg)) {
RCLog.d("pong[%s] <", id);
if (response.isNull("id")) {
return new DDPClientCallback.Ping(client, null);
} else {
String _id = response.optString("id");
if (id.equals(_id)) {
return new DDPClientCallback.Ping(client, _id);
} else {
return new DDPClientCallback.Ping.UnMatched(client, _id);
}
}
}
// if we receive anything other than a pong throw an exception
throw new DDPClientCallback.RPC.Error(client, id, response);
}).firstElement();
} else {
return Maybe.error(new DDPClientCallback.Closed(client));
/* package */ Maybe<DDPClientCallback.Base> ping(@Nullable final String id) {
final boolean requested = (TextUtils.isEmpty(id)) ?
sendMessage("ping", null) :
sendMessage("ping", json -> json.put("id", id));
if (requested) {
return flowable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
.timeout(8, TimeUnit.SECONDS)
.map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
.map(DDPClientImpl::toJson)
.filter(response -> "pong".equalsIgnoreCase(extractMsg(response)))
.doOnError(error -> {
RCLog.e(error, "Heartbeat ping[%s] xxx failed xxx", id);
})
.map(response -> {
String msg = extractMsg(response);
if ("pong".equals(msg)) {
RCLog.d("pong[%s] <", id);
if (response.isNull("id")) {
return new DDPClientCallback.Ping(client, null);
} else {
String _id = response.optString("id");
if (id.equals(_id)) {
return new DDPClientCallback.Ping(client, _id);
} else {
return new DDPClientCallback.Ping.UnMatched(client, _id);
}
}
}
// if we receive anything other than a pong throw an exception
throw new DDPClientCallback.RPC.Error(client, id, response);
}).firstElement();
} else {
return Maybe.error(new DDPClientCallback.Closed(client));
}
}
}
/* package */void ping(final TaskCompletionSource<DDPClientCallback.Ping> task,
@Nullable final String id) {
final boolean requested = (TextUtils.isEmpty(id)) ?
sendMessage("ping", null) :
sendMessage("ping", json -> json.put("id", id));
if (requested) {
CompositeDisposable disposables = new CompositeDisposable();
disposables.add(
flowable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
.timeout(8, TimeUnit.SECONDS)
.map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
.map(DDPClientImpl::toJson)
.subscribe(
response -> {
String msg = extractMsg(response);
if ("pong".equals(msg)) {
if (response.isNull("id")) {
task.setResult(new DDPClientCallback.Ping(client, null));
} else {
String _id = response.optString("id");
if (id.equals(_id)) {
task.setResult(new DDPClientCallback.Ping(client, id));
/* package */void ping(final TaskCompletionSource<DDPClientCallback.Ping> task,
@Nullable final String id) {
final boolean requested = (TextUtils.isEmpty(id)) ?
sendMessage("ping", null) :
sendMessage("ping", json -> json.put("id", id));
if (requested) {
CompositeDisposable disposables = new CompositeDisposable();
disposables.add(
flowable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
.timeout(8, TimeUnit.SECONDS)
.map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
.map(DDPClientImpl::toJson)
.subscribe(
response -> {
String msg = extractMsg(response);
if ("pong".equals(msg)) {
if (response.isNull("id")) {
task.setResult(new DDPClientCallback.Ping(client, null));
} else {
String _id = response.optString("id");
if (id.equals(_id)) {
task.setResult(new DDPClientCallback.Ping(client, id));
}
}
disposables.clear();
}
},
err -> {
if (err instanceof TimeoutException) {
task.trySetError(new Exception("Your connection seems off…"));
} else {
task.trySetError(new Exception("Ooops. Something's up!"));
}
}
}
disposables.clear();
}
},
err -> task.trySetError(new DDPClientCallback.Ping.Timeout(client))
)
);
addErrorCallback(disposables, task);
} else {
task.trySetError(new DDPClientCallback.Closed(client));
)
);
addErrorCallback(disposables, task);
} else {
task.trySetError(new DDPClientCallback.Closed(client));
}
}
}
/* package */ void sub(final TaskCompletionSource<DDPSubscription.Ready> task, String name,
JSONArray params, String id) {
final boolean requested =
sendMessage("sub", json -> json.put("id", id).put("name", name).put("params", params));
if (requested) {
CompositeDisposable disposables = new CompositeDisposable();
disposables.add(
flowable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
.map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
.map(DDPClientImpl::toJson)
.subscribe(
response -> {
String msg = extractMsg(response);
if ("ready".equals(msg) && !response.isNull("subs")) {
JSONArray ids = response.optJSONArray("subs");
for (int i = 0; i < ids.length(); i++) {
String _id = ids.optString(i);
if (id.equals(_id)) {
task.setResult(new DDPSubscription.Ready(client, id));
disposables.clear();
break;
}
}
} else if ("nosub".equals(msg) && !response.isNull("id") && !response.isNull(
"error")) {
String _id = response.optString("id");
if (id.equals(_id)) {
task.trySetError(new DDPSubscription.NoSub.Error(client, id,
response.optJSONObject("error")));
disposables.clear();
}
}
},
RCLog::e
)
);
addErrorCallback(disposables, task);
} else {
task.trySetError(new DDPClientCallback.Closed(client));
/* package */
void sub(final TaskCompletionSource<DDPSubscription.Ready> task, String name,
JSONArray params, String id) {
final boolean requested =
sendMessage("sub", json -> json.put("id", id).put("name", name).put("params", params));
if (requested) {
CompositeDisposable disposables = new CompositeDisposable();
disposables.add(
flowable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
.map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
.map(DDPClientImpl::toJson)
.subscribe(
response -> {
String msg = extractMsg(response);
if ("ready".equals(msg) && !response.isNull("subs")) {
JSONArray ids = response.optJSONArray("subs");
for (int i = 0; i < ids.length(); i++) {
String _id = ids.optString(i);
if (id.equals(_id)) {
task.setResult(new DDPSubscription.Ready(client, id));
disposables.clear();
break;
}
}
} else if ("nosub".equals(msg) && !response.isNull("id") && !response.isNull(
"error")) {
String _id = response.optString("id");
if (id.equals(_id)) {
task.trySetError(new DDPSubscription.NoSub.Error(client, id,
response.optJSONObject("error")));
disposables.clear();
}
}
},
err -> {
if (err instanceof TimeoutException) {
task.trySetError(new Exception("Your connection seems off…"));
} else {
task.trySetError(new Exception("Ooops. Something's up!"));
}
}
)
);
addErrorCallback(disposables, task);
} else {
task.trySetError(new DDPClientCallback.Closed(client));
}
}
}
/* package */ void unsub(final TaskCompletionSource<DDPSubscription.NoSub> task,
@Nullable final String id) {
/* package */ void unsub(final TaskCompletionSource<DDPSubscription.NoSub> task,
@Nullable final String id) {
final boolean requested = sendMessage("unsub", json -> json.put("id", id));
if (requested) {
CompositeDisposable disposables = new CompositeDisposable();
disposables.add(
flowable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
.map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
.map(DDPClientImpl::toJson)
.subscribe(
response -> {
String msg = extractMsg(response);
if ("nosub".equals(msg) && response.isNull("error") && !response.isNull("id")) {
String _id = response.optString("id");
if (id.equals(_id)) {
task.setResult(new DDPSubscription.NoSub(client, id));
disposables.clear();
}
}
},
err -> {
if (err instanceof TimeoutException) {
task.trySetError(new Exception("Your connection seems off…"));
} else {
task.trySetError(new Exception("Ooops. Something's up!"));
}
}
)
);
final boolean requested = sendMessage("unsub", json -> json.put("id", id));
addErrorCallback(disposables, task);
} else {
task.trySetError(new DDPClientCallback.Closed(client));
}
}
if (requested) {
CompositeDisposable disposables = new CompositeDisposable();
/* package */ void rpc(final TaskCompletionSource<DDPClientCallback.RPC> task, String method,
JSONArray params, String id, long timeoutMs) {
final boolean requested =
sendMessage("method",
json -> json.put("method", method).put("params", params).put("id", id));
if (requested) {
CompositeDisposable disposables = new CompositeDisposable();
disposables.add(
flowable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
.map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
.map(DDPClientImpl::toJson)
.timeout(timeoutMs, TimeUnit.MILLISECONDS)
.subscribe(
response -> {
String msg = extractMsg(response);
if ("result".equals(msg)) {
String _id = response.optString("id");
if (id.equals(_id)) {
if (!response.isNull("error")) {
task.trySetError(new DDPClientCallback.RPC.Error(client, id,
response.optJSONObject("error")));
} else {
String result = response.optString("result");
task.setResult(new DDPClientCallback.RPC(client, id, result));
}
disposables.clear();
}
}
},
err -> {
if (err instanceof TimeoutException) {
task.trySetError(new Exception("Your connection seems off…"));
} else {
task.trySetError(new Exception("Ooops. Something's up!"));
}
}
)
);
disposables.add(
flowable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
.map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
.map(DDPClientImpl::toJson)
.subscribe(
response -> {
String msg = extractMsg(response);
if ("nosub".equals(msg) && response.isNull("error") && !response.isNull("id")) {
String _id = response.optString("id");
if (id.equals(_id)) {
task.setResult(new DDPSubscription.NoSub(client, id));
disposables.clear();
}
}
},
err -> {
}
)
);
addErrorCallback(disposables, task);
} else {
task.trySetError(new DDPClientCallback.Closed(client));
addErrorCallback(disposables, task);
} else {
task.trySetError(new DDPClientCallback.Closed(client));
}
}
private void subscribeBaseListeners() {
if (disposables != null &&
disposables.size() > 0 && !disposables.isDisposed()) {
return;
}
disposables = new CompositeDisposable();
disposables.add(
flowable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
.map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
.map(DDPClientImpl::toJson)
.subscribe(
response -> {
String msg = extractMsg(response);
if ("ping".equals(msg)) {
if (response.isNull("id")) {
sendMessage("pong", null);
} else {
sendMessage("pong", json -> json.put("id", response.getString("id")));
}
}
},
RCLog::e
)
);
}
}
/* package */ void rpc(final TaskCompletionSource<DDPClientCallback.RPC> task, String method,
JSONArray params, String id, long timeoutMs) {
final boolean requested =
sendMessage("method",
json -> json.put("method", method).put("params", params).put("id", id));
if (requested) {
CompositeDisposable disposables = new CompositeDisposable();
disposables.add(
flowable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
.map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
.map(DDPClientImpl::toJson)
.timeout(timeoutMs, TimeUnit.MILLISECONDS)
.subscribe(
response -> {
/* package */ Flowable<DDPSubscription.Event> getDDPSubscription() {
String[] targetMsgs = {"added", "changed", "removed", "addedBefore", "movedBefore"};
return flowable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
.map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
.map(DDPClientImpl::toJson)
.filter(response -> {
String msg = extractMsg(response);
if ("result".equals(msg)) {
String _id = response.optString("id");
if (id.equals(_id)) {
if (!response.isNull("error")) {
task.trySetError(new DDPClientCallback.RPC.Error(client, id,
response.optJSONObject("error")));
} else {
String result = response.optString("result");
task.setResult(new DDPClientCallback.RPC(client, id, result));
for (String m : targetMsgs) {
if (m.equals(msg)) {
return true;
}
disposables.clear();
}
}
},
err -> {
if (err instanceof TimeoutException) {
task.trySetError(new DDPClientCallback.RPC.Timeout(client));
return false;
})
.map(response -> {
String msg = extractMsg(response);
if ("added".equals(msg)) {
return new DDPSubscription.Added(client, response.optString("collection"),
response.optString("id"),
response.isNull("fields") ? null : response.optJSONObject("fields"));
} else if ("addedBefore".equals(msg)) {
return new DDPSubscription.Added.Before(client, response.optString("collection"),
response.optString("id"),
response.isNull("fields") ? null : response.optJSONObject("fields"),
response.isNull("before") ? null : response.optString("before"));
} else if ("changed".equals(msg)) {
return new DDPSubscription.Changed(client, response.optString("collection"),
response.optString("id"),
response.isNull("fields") ? null : response.optJSONObject("fields"),
response.isNull("cleared") ? new JSONArray() : response.optJSONArray("before"));
} else if ("removed".equals(msg)) {
return new DDPSubscription.Removed(client, response.optString("collection"),
response.optString("id"));
} else if ("movedBefore".equals(msg)) {
return new DDPSubscription.MovedBefore(client, response.optString("collection"),
response.optString("id"),
response.isNull("before") ? null : response.optString("before"));
}
}
)
);
addErrorCallback(disposables, task);
} else {
task.trySetError(new DDPClientCallback.Closed(client));
return null;
});
}
}
private void subscribeBaseListeners() {
if (disposables != null &&
disposables.size() > 0 && !disposables.isDisposed()) {
return;
/* package */ void unsubscribeBaseListeners() {
if (disposables.size() > 0 || !disposables.isDisposed()) {
disposables.clear();
}
}
disposables = new CompositeDisposable();
disposables.add(
flowable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
.map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
.map(DDPClientImpl::toJson)
.subscribe(
response -> {
String msg = extractMsg(response);
if ("ping".equals(msg)) {
if (response.isNull("id")) {
sendMessage("pong", null);
} else {
sendMessage("pong", json -> json.put("id", response.getString("id")));
}
}
},
RCLog::e
)
);
}
/* package */ Flowable<DDPSubscription.Event> getDDPSubscription() {
String[] targetMsgs = {"added", "changed", "removed", "addedBefore", "movedBefore"};
return flowable.filter(callback -> callback instanceof RxWebSocketCallback.Message)
.map(callback -> ((RxWebSocketCallback.Message) callback).responseBodyString)
.map(DDPClientImpl::toJson)
.filter(response -> {
String msg = extractMsg(response);
for (String m : targetMsgs) {
if (m.equals(msg)) {
return true;
}
}
return false;
})
.map(response -> {
String msg = extractMsg(response);
if ("added".equals(msg)) {
return new DDPSubscription.Added(client, response.optString("collection"),
response.optString("id"),
response.isNull("fields") ? null : response.optJSONObject("fields"));
} else if ("addedBefore".equals(msg)) {
return new DDPSubscription.Added.Before(client, response.optString("collection"),
response.optString("id"),
response.isNull("fields") ? null : response.optJSONObject("fields"),
response.isNull("before") ? null : response.optString("before"));
} else if ("changed".equals(msg)) {
return new DDPSubscription.Changed(client, response.optString("collection"),
response.optString("id"),
response.isNull("fields") ? null : response.optJSONObject("fields"),
response.isNull("cleared") ? new JSONArray() : response.optJSONArray("before"));
} else if ("removed".equals(msg)) {
return new DDPSubscription.Removed(client, response.optString("collection"),
response.optString("id"));
} else if ("movedBefore".equals(msg)) {
return new DDPSubscription.MovedBefore(client, response.optString("collection"),
response.optString("id"),
response.isNull("before") ? null : response.optString("before"));
}
return null;
/* package */ Task<RxWebSocketCallback.Close> getOnCloseCallback() {
TaskCompletionSource<RxWebSocketCallback.Close> task = new TaskCompletionSource<>();
flowable.filter(callback -> callback instanceof RxWebSocketCallback.Close)
.cast(RxWebSocketCallback.Close.class)
.subscribe(
task::setResult,
err -> setTaskError(task, err)
);
return task.getTask().onSuccessTask(_task -> {
unsubscribeBaseListeners();
return _task;
});
}
}
/* package */ void unsubscribeBaseListeners() {
if (disposables.size() > 0 || !disposables.isDisposed()) {
disposables.clear();
private boolean sendMessage(String msg, @Nullable JSONBuilder json) {
try {
JSONObject origJson = new JSONObject().put("msg", msg);
String msg2 = (json == null ? origJson : json.create(origJson)).toString();
return websocket.sendText(msg2);
} catch (Exception e) {
RCLog.e(e);
return false;
}
}
}
/* package */ Task<RxWebSocketCallback.Close> getOnCloseCallback() {
TaskCompletionSource<RxWebSocketCallback.Close> task = new TaskCompletionSource<>();
private void sendMessage(String msg, @Nullable JSONBuilder json,
TaskCompletionSource<?> taskForSetError) {
if (!sendMessage(msg, json) && taskForSetError != null) {
taskForSetError.trySetError(new DDPClientCallback.Closed(client));
}
}
flowable.filter(callback -> callback instanceof RxWebSocketCallback.Close)
.cast(RxWebSocketCallback.Close.class)
.subscribe(
task::setResult,
err -> setTaskError(task, err)
private void addErrorCallback(CompositeDisposable disposables, TaskCompletionSource<?> task) {
disposables.add(
flowable.subscribe(
base -> {
if (base instanceof RxWebSocketCallback.Close) {
task.trySetError(new Exception(((RxWebSocketCallback.Close) base).reason));
}
},
err -> {
setTaskError(task, new Exception(err));
disposables.clear();
}
)
);
return task.getTask().onSuccessTask(_task -> {
unsubscribeBaseListeners();
return _task;
});
}
private boolean sendMessage(String msg, @Nullable JSONBuilder json) {
try {
JSONObject origJson = new JSONObject().put("msg", msg);
String msg2 = (json == null ? origJson : json.create(origJson)).toString();
return websocket.sendText(msg2);
} catch (Exception e) {
RCLog.e(e);
return false;
}
}
private void sendMessage(String msg, @Nullable JSONBuilder json,
TaskCompletionSource<?> taskForSetError) {
if (!sendMessage(msg, json) && taskForSetError != null) {
taskForSetError.trySetError(new DDPClientCallback.Closed(client));
}
}
private void addErrorCallback(CompositeDisposable disposables, TaskCompletionSource<?> task) {
disposables.add(
flowable.subscribe(
base -> {
if (base instanceof RxWebSocketCallback.Close) {
task.trySetError(new Exception(((RxWebSocketCallback.Close) base).reason));
}
},
err -> {
setTaskError(task, new Exception(err));
disposables.clear();
}
)
);
}
public void close(int code, String reason) {
try {
websocket.close(code, reason);
} catch (Exception e) {
RCLog.e(e);
public void close(int code, String reason) {
try {
websocket.close(code, reason);
} catch (Exception e) {
RCLog.e(e);
}
}
}
private void setTaskError(TaskCompletionSource task, Throwable throwable) {
if (task.getTask().isCompleted()) {
return;
private void setTaskError(TaskCompletionSource task, Throwable throwable) {
if (task.getTask().isCompleted()) {
return;
}
if (throwable instanceof Exception) {
task.setError((Exception) throwable);
} else {
task.setError(new Exception(throwable));
}
}
if (throwable instanceof Exception) {
task.setError((Exception) throwable);
} else {
task.setError(new Exception(throwable));
}
}
private interface JSONBuilder {
@NonNull
JSONObject create(JSONObject root) throws JSONException;
}
private interface JSONBuilder {
@NonNull
JSONObject create(JSONObject root) throws JSONException;
}
}
app/src/main/java/chat/rocket/android/ConnectionStatusManager.kt
View file @ abe34222
......@@ -11,6 +11,10 @@ object ConnectionStatusManager {
}
private const val DEBUG = false
private val DEFAULT_CALLBACK = object : TransitionCallback {
override fun onTransitioned(success: Boolean) {
}
}
private val stateMachine: EnumStateMachine<State>
init {
......@@ -28,23 +32,46 @@ object ConnectionStatusManager {
fun currentState() = stateMachine.currentState()
@Synchronized
fun setOnline(callback: TransitionCallback) {
fun setOnline(callback: TransitionCallback = DEFAULT_CALLBACK) {
KeepAliveJob.cancelPendingJobRequests()
tryTransitionTo(State.ONLINE, callback)
}
@Synchronized
fun setConnecting(callback: TransitionCallback) {
fun setOnline() {
KeepAliveJob.cancelPendingJobRequests()
tryTransitionTo(State.ONLINE, DEFAULT_CALLBACK)
}
@Synchronized
fun setConnecting(callback: TransitionCallback = DEFAULT_CALLBACK) {
KeepAliveJob.cancelPendingJobRequests()
tryTransitionTo(State.CONNECTING, callback)
}
@Synchronized
fun setConnectionError(callback: TransitionCallback) {
fun setConnecting() {
KeepAliveJob.cancelPendingJobRequests()
tryTransitionTo(State.CONNECTING, DEFAULT_CALLBACK)
}
@Synchronized
fun setConnectionError(callback: TransitionCallback = DEFAULT_CALLBACK) {
KeepAliveJob.schedule()
tryTransitionTo(State.OFFLINE, callback)
}
@Synchronized
fun setConnectionError() {
KeepAliveJob.schedule()
tryTransitionTo(State.OFFLINE, DEFAULT_CALLBACK)
}
@Synchronized
fun setOffline() {
stateMachine.reset()
}
private fun tryTransitionTo(newState: State, callback: TransitionCallback) {
try {
stateMachine.transition(newState)
......
app/src/main/java/chat/rocket/android/RocketChatCache.kt
View file @ abe34222
......@@ -72,7 +72,7 @@ object RocketChatCache {
return getString(KEY_SELECTED_SERVER_HOSTNAME, null)
}
fun setSelectedRoomId(roomId: String) {
fun setSelectedRoomId(roomId: String?) {
try {
val jsonObject = getSelectedRoomIdJsonObject()
jsonObject.put(getSelectedServerHostname(), roomId)
......@@ -81,7 +81,6 @@ object RocketChatCache {
RCLog.e(e)
Logger.report(e)
}
}
@Throws(JSONException::class)
......@@ -234,10 +233,11 @@ object RocketChatCache {
fun clearSelectedHostnameReferences() {
val hostname = getSelectedServerHostname()
if (hostname != null) {
setString(KEY_OPENED_ROOMS, null)
removeSiteName(hostname)
removeHostname(hostname)
removeSiteUrl(hostname)
setSelectedServerHostname(null)
setSelectedServerHostname(getFirstLoggedHostnameIfAny())
}
}
......@@ -308,7 +308,7 @@ object RocketChatCache {
fun setSessionToken(sessionToken: String) {
val selectedServerHostname = getSelectedServerHostname() ?:
throw IllegalStateException("Trying to set sessionToken to null hostname")
val sessions = getSessionToken()
val sessions = getString(KEY_SESSION_TOKEN, null)
try {
val jsonObject = if (sessions == null) JSONObject() else JSONObject(sessions)
jsonObject.put(selectedServerHostname, sessionToken)
......
app/src/main/java/chat/rocket/android/api/MethodCallHelper.java
View file @ abe34222
......@@ -118,7 +118,7 @@ public class MethodCallHelper {
} else if (exception instanceof DDPClientCallback.RPC.Timeout) {
return Task.forError(new MethodCall.Timeout());
} else if (exception instanceof DDPClientCallback.Closed) {
return Task.forError(new Exception("Oops, your connection seems off..."));
return Task.forError(new Exception(exception.getMessage()));
} else {
return Task.forError(exception);
}
......
app/src/main/java/chat/rocket/android/fragment/server_config/LoginPresenter.kt
View file @ abe34222
......@@ -17,6 +17,7 @@ import chat.rocket.core.models.PublicSetting
import chat.rocket.core.repositories.LoginServiceConfigurationRepository
import chat.rocket.core.repositories.PublicSettingRepository
import com.hadisatrio.optional.Optional
import io.reactivex.Completable
import io.reactivex.android.schedulers.AndroidSchedulers
import io.reactivex.rxkotlin.subscribeBy
......@@ -74,25 +75,34 @@ class LoginPresenter(private val loginServiceConfigurationRepository: LoginServi
}
private fun doLogin(username: String, password: String, optional: Optional<PublicSetting>) {
call(username, password, optional)
.continueWith(object : Continuation<Void, Any?> {
override fun then(task: Task<Void>?): Any? {
if (task != null && task.isFaulted()) {
view.hideLoader()
addSubscription(
Completable.create {
call(username, password, optional)
.continueWith(object : Continuation<Void, Any?> {
override fun then(task: Task<Void>?): Any? {
if (task != null && task.isFaulted()) {
view.hideLoader()
val error = task.getError()
val error = task.getError()
error?.let {
if (error is TwoStepAuthException) {
view.showTwoStepAuth()
} else {
view.showError(error.message)
error?.let {
if (error is TwoStepAuthException) {
view.showTwoStepAuth()
} else {
view.showError(error.message)
}
}
return Completable.complete()
}
return null
}
}
}, Task.UI_THREAD_EXECUTOR)
}.subscribeBy(
onError = {
view.showError(it.message)
}
return null
}
}, Task.UI_THREAD_EXECUTOR)
)
)
}
private fun call(username: String, password: String, optional: Optional<PublicSetting>): Task<Void> {
......
app/src/main/java/chat/rocket/android/service/RealmBasedConnectivityManager.java
View file @ abe34222
......@@ -14,6 +14,7 @@ import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import chat.rocket.android.ConnectionStatusManager;
import chat.rocket.android.RocketChatCache;
import chat.rocket.android.helper.RxHelper;
import chat.rocket.android.log.RCLog;
......@@ -260,6 +261,7 @@ import io.reactivex.subjects.BehaviorSubject;
private Single<Boolean> connectToServer(String hostname) {
return Single.defer(() -> {
if (!serverConnectivityList.containsKey(hostname)) {
ConnectionStatusManager.INSTANCE.setConnectionError();
return Single.error(new IllegalArgumentException("hostname not found"));
}
......@@ -270,8 +272,10 @@ import io.reactivex.subjects.BehaviorSubject;
}
if (serviceInterface != null) {
ConnectionStatusManager.INSTANCE.setConnecting();
return serviceInterface.ensureConnectionToServer(hostname);
} else {
ConnectionStatusManager.INSTANCE.setConnectionError();
return Single.error(new ThreadLooperNotPreparedException("not prepared"));
}
});
......
app/src/main/java/chat/rocket/android/service/RocketChatService.java
View file @ abe34222
......@@ -38,7 +38,8 @@ public class RocketChatService extends Service implements ConnectivityServiceInt
/**
* ensure RocketChatService alive.
*/
/*package*/static void keepAlive(Context context) {
/*package*/
static void keepAlive(Context context) {
context.startService(new Intent(context, RocketChatService.class));
}
......@@ -99,31 +100,29 @@ public class RocketChatService extends Service implements ConnectivityServiceInt
return Single.defer(() -> {
webSocketThreadLock.acquire();
int connectivityState = ConnectivityManager.getInstance(getApplicationContext()).getConnectivityState(hostname);
boolean isDisconnected = connectivityState < ServerConnectivity.STATE_CONNECTED ;
boolean isDisconnected = connectivityState < ServerConnectivity.STATE_CONNECTED;
if (currentWebSocketThread != null && existsThreadForHostname(hostname) && !isDisconnected) {
webSocketThreadLock.release();
return Single.just(currentWebSocketThread);
}
if (currentWebSocketThread != null) {
if (isDisconnected) {
return currentWebSocketThread.terminate(true)
.doAfterTerminate(() -> currentWebSocketThread = null)
.flatMap(terminated ->
RocketChatWebSocketThread.getStarted(getApplicationContext(), hostname)
.doOnSuccess(thread -> {
currentWebSocketThread = thread;
webSocketThreadLock.release();
})
.doOnError(throwable -> {
currentWebSocketThread = null;
RCLog.e(throwable);
Logger.INSTANCE.report(throwable);
webSocketThreadLock.release();
})
);
}
return Single.just(currentWebSocketThread);
boolean hasFailed = existsThreadForHostname(hostname);
return currentWebSocketThread.terminate(hasFailed)
.doAfterTerminate(() -> currentWebSocketThread = null)
.flatMap(terminated ->
RocketChatWebSocketThread.getStarted(getApplicationContext(), hostname)
.doOnSuccess(thread -> {
currentWebSocketThread = thread;
webSocketThreadLock.release();
})
.doOnError(throwable -> {
currentWebSocketThread = null;
RCLog.e(throwable);
Logger.INSTANCE.report(throwable);
webSocketThreadLock.release();
})
);
}
return RocketChatWebSocketThread.getStarted(getApplicationContext(), hostname)
......
app/src/main/java/chat/rocket/android/service/RocketChatWebSocketThread.java
View file @ abe34222
......@@ -14,6 +14,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import bolts.Task;
import chat.rocket.android.ConnectionStatusManager;
import chat.rocket.android.api.MethodCallHelper;
import chat.rocket.android.helper.LogIfError;
import chat.rocket.android.helper.RxHelper;
......@@ -123,26 +124,26 @@ public class RocketChatWebSocketThread extends HandlerThread {
}
/**
* terminate WebSocket thread.
* Terminate WebSocket thread. If {@code hasFailed} is {@code true} it means that a connection was
* in progress but failed and got offline. If it's {@code false} means that the user explicitly
* disconnected from server either by logging out or by means of changing servers.
*
* @param isDisconnected {@code true} If we're trying to terminate a disconnected a websocket
* thread which means it has failed.
* @param hasFailed {@code true} if the termination is due to a network error, otherwise
* return false
*/
@DebugLog
/* package */ Single<Boolean> terminate(boolean isDisconnected) {
/* package */ Single<Boolean> terminate(boolean hasFailed) {
if (isAlive()) {
return Single.create(emitter -> {
new Handler(getLooper()).post(() -> {
RCLog.d("thread %s: terminated()", Thread.currentThread().getId());
int reason = (isDisconnected) ?
DDPClient.REASON_NETWORK_ERROR : DDPClient.REASON_CLOSED_BY_USER;
unregisterListenersAndClose(reason);
connectivityManager.notifyConnectionLost(hostname,
isDisconnected ? DDPClient.REASON_NETWORK_ERROR : DDPClient.REASON_CLOSED_BY_USER);
RocketChatWebSocketThread.super.quit();
emitter.onSuccess(true);
});
});
return Single.create(emitter -> new Handler(getLooper()).post(() -> {
RCLog.d("thread %s: terminated()", Thread.currentThread().getId());
int reason = (hasFailed) ?
DDPClient.REASON_NETWORK_ERROR : DDPClient.REASON_CLOSED_BY_USER;
unregisterListenersAndClose(reason);
connectivityManager.notifyConnectionLost(hostname, reason);
RocketChatWebSocketThread.super.quit();
ConnectionStatusManager.INSTANCE.setOffline();
emitter.onSuccess(true);
}));
} else {
connectivityManager.notifyConnectionLost(hostname,
DDPClient.REASON_NETWORK_ERROR);
......@@ -166,7 +167,7 @@ public class RocketChatWebSocketThread extends HandlerThread {
@DebugLog
/* package */ Single<Boolean> keepAlive() {
return checkIfConnectionAlive()
.flatMap(alive -> alive ? Single.just(true) : connectWithExponentialBackoff());
.flatMap(alive -> connectWithExponentialBackoff());
}
@DebugLog
......@@ -227,7 +228,7 @@ public class RocketChatWebSocketThread extends HandlerThread {
.onSuccessTask(task -> {
final String newSession = task.getResult().session;
connectivityManager.notifyConnectionEstablished(hostname, newSession);
// handling WebSocket#onClose() callback.
task.getResult().client.getOnCloseCallback().onSuccess(_task -> {
RxWebSocketCallback.Close result = _task.getResult();
if (result.code == DDPClient.REASON_NETWORK_ERROR) {
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment

Technounit-GROUP 2023