Skip to content

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

Refactor indentation

parent 184cfda5
Hide whitespace changes
Inline Side-by-side
Showing with 583 additions and 585 deletions
app/src/main/java/chat/rocket/android/activity/MainPresenter.java
View file @ 87461e2a
...@@ -33,227 +33,227 @@ import io.reactivex.android.schedulers.AndroidSchedulers; ...@@ -33,227 +33,227 @@ import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable; import io.reactivex.disposables.Disposable;
public class MainPresenter extends BasePresenter<MainContract.View> public class MainPresenter extends BasePresenter<MainContract.View>
implements MainContract.Presenter { implements MainContract.Presenter {
private final CanCreateRoomInteractor canCreateRoomInteractor; private final CanCreateRoomInteractor canCreateRoomInteractor;
private final RoomInteractor roomInteractor; private final RoomInteractor roomInteractor;
private final SessionInteractor sessionInteractor; private final SessionInteractor sessionInteractor;
private final MethodCallHelper methodCallHelper; private final MethodCallHelper methodCallHelper;
private final ConnectivityManagerApi connectivityManagerApi; private final ConnectivityManagerApi connectivityManagerApi;
private final RocketChatCache rocketChatCache; private final RocketChatCache rocketChatCache;
private final PublicSettingRepository publicSettingRepository; private final PublicSettingRepository publicSettingRepository;
public MainPresenter(RoomInteractor roomInteractor, public MainPresenter(RoomInteractor roomInteractor,
CanCreateRoomInteractor canCreateRoomInteractor, CanCreateRoomInteractor canCreateRoomInteractor,
SessionInteractor sessionInteractor, SessionInteractor sessionInteractor,
MethodCallHelper methodCallHelper, MethodCallHelper methodCallHelper,
ConnectivityManagerApi connectivityManagerApi, ConnectivityManagerApi connectivityManagerApi,
RocketChatCache rocketChatCache, PublicSettingRepository publicSettingRepository) { RocketChatCache rocketChatCache, PublicSettingRepository publicSettingRepository) {
this.roomInteractor = roomInteractor; this.roomInteractor = roomInteractor;
this.canCreateRoomInteractor = canCreateRoomInteractor; this.canCreateRoomInteractor = canCreateRoomInteractor;
this.sessionInteractor = sessionInteractor; this.sessionInteractor = sessionInteractor;
this.methodCallHelper = methodCallHelper; this.methodCallHelper = methodCallHelper;
this.connectivityManagerApi = connectivityManagerApi; this.connectivityManagerApi = connectivityManagerApi;
this.rocketChatCache = rocketChatCache; this.rocketChatCache = rocketChatCache;
this.publicSettingRepository = publicSettingRepository; this.publicSettingRepository = publicSettingRepository;
}
@Override
public void bindViewOnly(@NonNull MainContract.View view) {
super.bindView(view);
subscribeToUnreadCount();
subscribeToSession();
setUserOnline();
}
@Override
public void loadSignedInServers(@NonNull String hostname) {
final Disposable disposable = publicSettingRepository.getById(PublicSettingsConstants.Assets.LOGO)
.zipWith(publicSettingRepository.getById(PublicSettingsConstants.General.SITE_NAME), Pair::new)
.map(this::getLogoAndSiteNamePair)
.map(settings -> getServerList(hostname, settings))
.subscribeOn(AndroidSchedulers.from(BackgroundLooper.get()))
.observeOn(AndroidSchedulers.mainThread())
.subscribe(
view::showSignedInServers,
RCLog::e
);
addSubscription(disposable);
}
@Override
public void bindView(@NonNull MainContract.View view) {
super.bindView(view);
if (shouldLaunchAddServerActivity()) {
view.showAddServerScreen();
return;
} }
openRoom(); @Override
public void bindViewOnly(@NonNull MainContract.View view) {
subscribeToNetworkChanges(); super.bindView(view);
subscribeToUnreadCount(); subscribeToUnreadCount();
subscribeToSession(); subscribeToSession();
setUserOnline(); setUserOnline();
}
@Override
public void release() {
setUserAway();
super.release();
}
@Override
public void onOpenRoom(String hostname, String roomId) {
final Disposable subscription = canCreateRoomInteractor.canCreate(roomId)
.subscribeOn(AndroidSchedulers.from(BackgroundLooper.get()))
.observeOn(AndroidSchedulers.mainThread())
.subscribe(
allowed -> {
if (allowed) {
view.showRoom(hostname, roomId);
} else {
view.showHome();
}
},
Logger::report
);
addSubscription(subscription);
}
@Override
public void onRetryLogin() {
final Disposable subscription = sessionInteractor.retryLogin()
.subscribe();
addSubscription(subscription);
}
@Override
public void beforeLogoutCleanUp() {
clearSubscriptions();
}
private Pair<String, String> getLogoAndSiteNamePair(Pair<Optional<PublicSetting>, Optional<PublicSetting>> settingsPair) {
String logoUrl = "";
String siteName = "";
if (settingsPair.first.isPresent()) {
logoUrl = settingsPair.first.get().getValue();
}
if (settingsPair.second.isPresent()) {
siteName = settingsPair.second.get().getValue();
}
return new Pair<>(logoUrl, siteName);
}
private List<Pair<String, Pair<String, String>>> getServerList(String hostname, Pair<String, String> serverInfoPair) throws JSONException {
JSONObject jsonObject = new JSONObject(serverInfoPair.first);
String logoUrl = (jsonObject.has("url")) ?
jsonObject.optString("url") : jsonObject.optString("defaultUrl");
String siteName = serverInfoPair.second;
rocketChatCache.addHostname(hostname.toLowerCase(), logoUrl, siteName);
return rocketChatCache.getServerList();
}
private void openRoom() {
String hostname = rocketChatCache.getSelectedServerHostname();
String roomId = rocketChatCache.getSelectedRoomId();
if (roomId == null || roomId.length() == 0) {
view.showHome();
return;
} }
onOpenRoom(hostname, roomId); @Override
} public void loadSignedInServers(@NonNull String hostname) {
final Disposable disposable = publicSettingRepository.getById(PublicSettingsConstants.Assets.LOGO)
private void subscribeToUnreadCount() { .zipWith(publicSettingRepository.getById(PublicSettingsConstants.General.SITE_NAME), Pair::new)
final Disposable subscription = Flowable.combineLatest( .map(this::getLogoAndSiteNamePair)
roomInteractor.getTotalUnreadRoomsCount(), .map(settings -> getServerList(hostname, settings))
roomInteractor.getTotalUnreadMentionsCount(), .subscribeOn(AndroidSchedulers.from(BackgroundLooper.get()))
(Pair::new) .observeOn(AndroidSchedulers.mainThread())
) .subscribe(
.subscribeOn(AndroidSchedulers.from(BackgroundLooper.get())) view::showSignedInServers,
.observeOn(AndroidSchedulers.mainThread()) RCLog::e
.subscribe( );
pair -> view.showUnreadCount(pair.first, pair.second),
Logger::report addSubscription(disposable);
); }
addSubscription(subscription); @Override
} public void bindView(@NonNull MainContract.View view) {
super.bindView(view);
private void subscribeToSession() {
final Disposable subscription = sessionInteractor.getDefault() if (shouldLaunchAddServerActivity()) {
.subscribeOn(AndroidSchedulers.from(BackgroundLooper.get())) view.showAddServerScreen();
.observeOn(AndroidSchedulers.mainThread()) return;
.subscribe( }
sessionOptional -> {
Session session = sessionOptional.orNull(); openRoom();
if (session == null || session.getToken() == null) {
view.showLoginScreen(); subscribeToNetworkChanges();
return; subscribeToUnreadCount();
} subscribeToSession();
setUserOnline();
String error = session.getError(); }
if (error != null && error.length() != 0) {
view.showConnectionError(); @Override
return; public void release() {
} setUserAway();
if (!session.isTokenVerified()) { super.release();
view.showConnecting(); }
return;
} @Override
public void onOpenRoom(String hostname, String roomId) {
view.showConnectionOk(); final Disposable subscription = canCreateRoomInteractor.canCreate(roomId)
}, .subscribeOn(AndroidSchedulers.from(BackgroundLooper.get()))
Logger::report .observeOn(AndroidSchedulers.mainThread())
); .subscribe(
allowed -> {
addSubscription(subscription); if (allowed) {
} view.showRoom(hostname, roomId);
} else {
private void subscribeToNetworkChanges() { view.showHome();
Disposable disposable = connectivityManagerApi.getServerConnectivityAsObservable() }
.distinctUntilChanged() },
.observeOn(AndroidSchedulers.mainThread()) Logger::report
.subscribe( );
connectivity -> {
if (connectivity.state == ServerConnectivity.STATE_CONNECTED) { addSubscription(subscription);
view.showConnectionOk(); }
view.refreshRoom();
} else if (connectivity.state == ServerConnectivity.STATE_DISCONNECTED) { @Override
if (connectivity.code == DDPClient.REASON_NETWORK_ERROR) { public void onRetryLogin() {
view.showConnectionError(); final Disposable subscription = sessionInteractor.retryLogin()
} else { .subscribe();
view.showConnectionOk();
} addSubscription(subscription);
} else { }
view.showConnecting();
} @Override
}, public void beforeLogoutCleanUp() {
Logger::report clearSubscriptions();
); }
addSubscription(disposable); private Pair<String, String> getLogoAndSiteNamePair(Pair<Optional<PublicSetting>, Optional<PublicSetting>> settingsPair) {
} String logoUrl = "";
String siteName = "";
private void setUserOnline() { if (settingsPair.first.isPresent()) {
methodCallHelper.setUserPresence(User.STATUS_ONLINE) logoUrl = settingsPair.first.get().getValue();
.continueWith(new LogIfError()); }
} if (settingsPair.second.isPresent()) {
siteName = settingsPair.second.get().getValue();
private void setUserAway() { }
methodCallHelper.setUserPresence(User.STATUS_AWAY) return new Pair<>(logoUrl, siteName);
.continueWith(new LogIfError()); }
}
private List<Pair<String, Pair<String, String>>> getServerList(String hostname, Pair<String, String> serverInfoPair) throws JSONException {
private boolean shouldLaunchAddServerActivity() { JSONObject jsonObject = new JSONObject(serverInfoPair.first);
return connectivityManagerApi.getServerList().isEmpty(); String logoUrl = (jsonObject.has("url")) ?
} jsonObject.optString("url") : jsonObject.optString("defaultUrl");
String siteName = serverInfoPair.second;
rocketChatCache.addHostname(hostname.toLowerCase(), logoUrl, siteName);
return rocketChatCache.getServerList();
}
private void openRoom() {
String hostname = rocketChatCache.getSelectedServerHostname();
String roomId = rocketChatCache.getSelectedRoomId();
if (roomId == null || roomId.length() == 0) {
view.showHome();
return;
}
onOpenRoom(hostname, roomId);
}
private void subscribeToUnreadCount() {
final Disposable subscription = Flowable.combineLatest(
roomInteractor.getTotalUnreadRoomsCount(),
roomInteractor.getTotalUnreadMentionsCount(),
(Pair::new)
)
.subscribeOn(AndroidSchedulers.from(BackgroundLooper.get()))
.observeOn(AndroidSchedulers.mainThread())
.subscribe(
pair -> view.showUnreadCount(pair.first, pair.second),
Logger::report
);
addSubscription(subscription);
}
private void subscribeToSession() {
final Disposable subscription = sessionInteractor.getDefault()
.subscribeOn(AndroidSchedulers.from(BackgroundLooper.get()))
.observeOn(AndroidSchedulers.mainThread())
.subscribe(
sessionOptional -> {
Session session = sessionOptional.orNull();
if (session == null || session.getToken() == null) {
view.showLoginScreen();
return;
}
String error = session.getError();
if (error != null && error.length() != 0) {
view.showConnectionError();
return;
}
if (!session.isTokenVerified()) {
view.showConnecting();
return;
}
view.showConnectionOk();
},
Logger::report
);
addSubscription(subscription);
}
private void subscribeToNetworkChanges() {
Disposable disposable = connectivityManagerApi.getServerConnectivityAsObservable()
.distinctUntilChanged()
.observeOn(AndroidSchedulers.mainThread())
.subscribe(
connectivity -> {
if (connectivity.state == ServerConnectivity.STATE_CONNECTED) {
view.showConnectionOk();
view.refreshRoom();
} else if (connectivity.state == ServerConnectivity.STATE_DISCONNECTED) {
if (connectivity.code == DDPClient.REASON_NETWORK_ERROR) {
view.showConnectionError();
} else {
view.showConnectionOk();
}
} else {
view.showConnecting();
}
},
Logger::report
);
addSubscription(disposable);
}
private void setUserOnline() {
methodCallHelper.setUserPresence(User.STATUS_ONLINE)
.continueWith(new LogIfError());
}
private void setUserAway() {
methodCallHelper.setUserPresence(User.STATUS_AWAY)
.continueWith(new LogIfError());
}
private boolean shouldLaunchAddServerActivity() {
return connectivityManagerApi.getServerList().isEmpty();
}
} }
app/src/main/java/chat/rocket/android/service/ConnectivityManagerApi.java
View file @ 87461e2a
...@@ -13,19 +13,19 @@ import io.reactivex.Single; ...@@ -13,19 +13,19 @@ import io.reactivex.Single;
* interfaces used for Activity/Fragment and other UI-related logic. * interfaces used for Activity/Fragment and other UI-related logic.
*/ */
public interface ConnectivityManagerApi { public interface ConnectivityManagerApi {
void keepAliveServer(); void keepAliveServer();
void addOrUpdateServer(String hostname, @Nullable String name, boolean insecure); void addOrUpdateServer(String hostname, @Nullable String name, boolean insecure);
void removeServer(String hostname); void removeServer(String hostname);
Single<Boolean> connect(String hostname); Single<Boolean> connect(String hostname);
List<ServerInfo> getServerList(); List<ServerInfo> getServerList();
Flowable<ServerConnectivity> getServerConnectivityAsObservable(); Flowable<ServerConnectivity> getServerConnectivityAsObservable();
int getConnectivityState(@NonNull String hostname); int getConnectivityState(@NonNull String hostname);
void resetConnectivityStateList(); void resetConnectivityStateList();
} }
app/src/main/java/chat/rocket/android/service/RealmBasedConnectivityManager.java
View file @ 87461e2a
...@@ -23,264 +23,262 @@ import chat.rocket.persistence.realm.models.RealmBasedServerInfo; ...@@ -23,264 +23,262 @@ import chat.rocket.persistence.realm.models.RealmBasedServerInfo;
import hugo.weaving.DebugLog; import hugo.weaving.DebugLog;
import io.reactivex.BackpressureStrategy; import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable; import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.Single; import io.reactivex.Single;
import io.reactivex.schedulers.Schedulers; import io.reactivex.schedulers.Schedulers;
import io.reactivex.subjects.BehaviorSubject; import io.reactivex.subjects.BehaviorSubject;
import io.reactivex.subjects.PublishSubject;
/** /**
* Connectivity management implementation. * Connectivity management implementation.
*/ */
/*package*/ class RealmBasedConnectivityManager /*package*/ class RealmBasedConnectivityManager
implements ConnectivityManagerApi, ConnectivityManagerInternal { implements ConnectivityManagerApi, ConnectivityManagerInternal {
private volatile ConcurrentHashMap<String, Integer> serverConnectivityList = new ConcurrentHashMap<>(); private volatile ConcurrentHashMap<String, Integer> serverConnectivityList = new ConcurrentHashMap<>();
private volatile BehaviorSubject<ServerConnectivity> connectivitySubject = BehaviorSubject.createDefault(ServerConnectivity.CONNECTED); private volatile BehaviorSubject<ServerConnectivity> connectivitySubject = BehaviorSubject.createDefault(ServerConnectivity.CONNECTED);
private Context appContext; private Context appContext;
private final ServiceConnection serviceConnection = new ServiceConnection() { private final ServiceConnection serviceConnection = new ServiceConnection() {
@Override
public void onServiceConnected(ComponentName componentName, IBinder binder) {
serviceInterface = ((RocketChatService.LocalBinder) binder).getServiceInterface();
}
@Override
public void onServiceDisconnected(ComponentName componentName) {
serviceInterface = null;
}
};
private ConnectivityServiceInterface serviceInterface;
/*package*/ RealmBasedConnectivityManager setContext(Context appContext) {
this.appContext = appContext.getApplicationContext();
return this;
}
@Override
public void resetConnectivityStateList() {
serverConnectivityList.clear();
for (ServerInfo serverInfo : RealmBasedServerInfo.getServerInfoList()) {
serverConnectivityList.put(serverInfo.getHostname(), ServerConnectivity.STATE_DISCONNECTED);
}
}
@Override
public void keepAliveServer() {
RocketChatService.keepAlive(appContext);
if (serviceInterface == null) {
RocketChatService.bind(appContext, serviceConnection);
}
}
@SuppressLint("RxLeakedSubscription")
@DebugLog
@Override
public void ensureConnections() {
String hostname = new RocketChatCache(appContext).getSelectedServerHostname();
if (hostname == null) {
return;
}
connectToServerIfNeeded(hostname, true/* force connect */)
.subscribeOn(Schedulers.io())
.subscribe(_val -> {
}, error -> {
RCLog.e(error);
notifyConnectionLost(hostname, DDPClient.REASON_NETWORK_ERROR);
});
}
@SuppressLint("RxLeakedSubscription")
@Override
public void addOrUpdateServer(String hostname, @Nullable String name, boolean insecure) {
RealmBasedServerInfo.addOrUpdate(hostname, name, insecure);
if (!serverConnectivityList.containsKey(hostname)) {
serverConnectivityList.put(hostname, ServerConnectivity.STATE_DISCONNECTED);
}
connectToServerIfNeeded(hostname, false)
.subscribe(_val -> {
}, RCLog::e);
}
@SuppressLint("RxLeakedSubscription")
@Override
public void removeServer(String hostname) {
RealmBasedServerInfo.remove(hostname);
if (serverConnectivityList.containsKey(hostname)) {
disconnectFromServerIfNeeded(hostname)
.subscribe(_val -> {
}, RCLog::e);
}
}
@Override
public Single<Boolean> connect(String hostname) {
return connectToServerIfNeeded(hostname, false);
}
@Override @Override
public void onServiceConnected(ComponentName componentName, IBinder binder) { public List<ServerInfo> getServerList() {
serviceInterface = ((RocketChatService.LocalBinder) binder).getServiceInterface(); return RealmBasedServerInfo.getServerInfoList();
} }
@Override @Override
public void onServiceDisconnected(ComponentName componentName) { public ServerInfo getServerInfoForHost(String hostname) {
serviceInterface = null; return RealmBasedServerInfo.getServerInfoForHost(hostname);
} }
};
private ConnectivityServiceInterface serviceInterface;
private List<ServerConnectivity> getCurrentConnectivityList() {
ArrayList<ServerConnectivity> list = new ArrayList<>();
for (Map.Entry<String, Integer> entry : serverConnectivityList.entrySet()) {
list.add(new ServerConnectivity(entry.getKey(), entry.getValue()));
}
return list;
}
/*package*/ RealmBasedConnectivityManager setContext(Context appContext) { @DebugLog
this.appContext = appContext.getApplicationContext(); @Override
return this; public void notifyConnectionEstablished(String hostname, String session) {
} if (session != null) {
RealmBasedServerInfo.updateSession(hostname, session);
}
serverConnectivityList.put(hostname, ServerConnectivity.STATE_CONNECTED);
connectivitySubject.onNext(
new ServerConnectivity(hostname, ServerConnectivity.STATE_CONNECTED));
}
@Override @DebugLog
public void resetConnectivityStateList() { @Override
serverConnectivityList.clear(); public void notifyConnectionLost(String hostname, int code) {
for (ServerInfo serverInfo : RealmBasedServerInfo.getServerInfoList()) { serverConnectivityList.put(hostname, ServerConnectivity.STATE_DISCONNECTED);
serverConnectivityList.put(serverInfo.getHostname(), ServerConnectivity.STATE_DISCONNECTED); connectivitySubject.onNext(
new ServerConnectivity(hostname, ServerConnectivity.STATE_DISCONNECTED, code));
} }
}
@Override @DebugLog
public void keepAliveServer() { @Override
RocketChatService.keepAlive(appContext); public void notifyConnecting(String hostname) {
if (serviceInterface == null) { serverConnectivityList.put(hostname, ServerConnectivity.STATE_CONNECTING);
RocketChatService.bind(appContext, serviceConnection); connectivitySubject.onNext(
new ServerConnectivity(hostname, ServerConnectivity.STATE_CONNECTING));
} }
}
@Override
@SuppressLint("RxLeakedSubscription") public Flowable<ServerConnectivity> getServerConnectivityAsObservable() {
@DebugLog return connectivitySubject.toFlowable(BackpressureStrategy.LATEST);
@Override
public void ensureConnections() {
String hostname = new RocketChatCache(appContext).getSelectedServerHostname();
if (hostname == null) {
return;
} }
connectToServerIfNeeded(hostname, true/* force connect */)
.subscribeOn(Schedulers.io()) @Override
.subscribe(_val -> { public int getConnectivityState(@NonNull String hostname) {
}, error -> { return serverConnectivityList.get(hostname);
RCLog.e(error);
notifyConnectionLost(hostname, DDPClient.REASON_NETWORK_ERROR);
});
}
@SuppressLint("RxLeakedSubscription")
@Override
public void addOrUpdateServer(String hostname, @Nullable String name, boolean insecure) {
RealmBasedServerInfo.addOrUpdate(hostname, name, insecure);
if (!serverConnectivityList.containsKey(hostname)) {
serverConnectivityList.put(hostname, ServerConnectivity.STATE_DISCONNECTED);
} }
connectToServerIfNeeded(hostname, false)
.subscribe(_val -> { @DebugLog
}, RCLog::e); private Single<Boolean> connectToServerIfNeeded(String hostname, boolean forceConnect) {
} return Single.defer(() -> {
Integer state = serverConnectivityList.get(hostname);
@SuppressLint("RxLeakedSubscription") if (state == null) {
@Override state = ServerConnectivity.STATE_DISCONNECTED;
public void removeServer(String hostname) { }
RealmBasedServerInfo.remove(hostname); final int connectivity = state;
if (serverConnectivityList.containsKey(hostname)) { if (!forceConnect && connectivity == ServerConnectivity.STATE_CONNECTED) {
disconnectFromServerIfNeeded(hostname) return Single.just(true);
.subscribe(_val -> { }
}, RCLog::e);
if (connectivity == ServerConnectivity.STATE_DISCONNECTING) {
return waitForDisconnected(hostname)
.flatMap(_val -> connectToServerIfNeeded(hostname, forceConnect));
}
if (connectivity == ServerConnectivity.STATE_DISCONNECTED) {
// notifyConnecting(hostname);
}
return connectToServer(hostname);
});
} }
}
private Single<Boolean> disconnectFromServerIfNeeded(String hostname) {
@Override return Single.defer(() -> {
public Single<Boolean> connect(String hostname) { final int connectivity = serverConnectivityList.get(hostname);
return connectToServerIfNeeded(hostname, false); if (connectivity == ServerConnectivity.STATE_DISCONNECTED) {
} return Single.just(true);
}
@Override
public List<ServerInfo> getServerList() { if (connectivity == ServerConnectivity.STATE_CONNECTING) {
return RealmBasedServerInfo.getServerInfoList(); return waitForConnected(hostname)
} .doOnError(err -> notifyConnectionLost(hostname, DDPClient.REASON_NETWORK_ERROR))
.flatMap(_val -> disconnectFromServerIfNeeded(hostname));
@Override }
public ServerInfo getServerInfoForHost(String hostname) {
return RealmBasedServerInfo.getServerInfoForHost(hostname); if (connectivity == ServerConnectivity.STATE_DISCONNECTING) {
} return waitForDisconnected(hostname);
}
private List<ServerConnectivity> getCurrentConnectivityList() {
ArrayList<ServerConnectivity> list = new ArrayList<>(); return disconnectFromServer(hostname)
for (Map.Entry<String, Integer> entry : serverConnectivityList.entrySet()) { .retryWhen(RxHelper.exponentialBackoff(1, 500, TimeUnit.MILLISECONDS));
list.add(new ServerConnectivity(entry.getKey(), entry.getValue())); });
} }
return list;
} @DebugLog
private Single<Boolean> waitForConnected(String hostname) {
@DebugLog return connectivitySubject
@Override .filter(serverConnectivity -> hostname.equals(serverConnectivity.hostname))
public void notifyConnectionEstablished(String hostname, String session) { .map(serverConnectivity -> serverConnectivity.state)
if (session != null) { .filter(state ->
RealmBasedServerInfo.updateSession(hostname, session); state == ServerConnectivity.STATE_CONNECTED
|| state == ServerConnectivity.STATE_DISCONNECTED)
.firstElement()
.toSingle()
.flatMap(state ->
state == ServerConnectivity.STATE_CONNECTED
? Single.just(true)
: Single.error(new ServerConnectivity.DisconnectedException()));
}
@DebugLog
private Single<Boolean> waitForDisconnected(String hostname) {
return connectivitySubject
.filter(serverConnectivity -> hostname.equals(serverConnectivity.hostname))
.map(serverConnectivity -> serverConnectivity.state)
.filter(state -> state == ServerConnectivity.STATE_DISCONNECTED)
.firstElement()
.toSingle()
.map(state -> true);
}
@DebugLog
private Single<Boolean> connectToServer(String hostname) {
return Single.defer(() -> {
if (!serverConnectivityList.containsKey(hostname)) {
return Single.error(new IllegalArgumentException("hostname not found"));
}
if (serverConnectivityList.get(hostname) != ServerConnectivity.STATE_CONNECTED) {
// Mark as CONNECTING except for the case [forceConnect && connected] because
// ensureConnectionToServer doesn't notify ConnectionEstablished/Lost is already connected.
// serverConnectivityList.put(hostname, ServerConnectivity.STATE_CONNECTING);
}
if (serviceInterface != null) {
return serviceInterface.ensureConnectionToServer(hostname);
} else {
return Single.error(new IllegalStateException("not prepared"));
}
});
}
private Single<Boolean> disconnectFromServer(String hostname) {
return Single.defer(() -> {
if (!serverConnectivityList.containsKey(hostname)) {
return Single.error(new IllegalArgumentException("hostname not found"));
}
serverConnectivityList.put(hostname, ServerConnectivity.STATE_DISCONNECTING);
if (serviceInterface != null) {
return serviceInterface.disconnectFromServer(hostname)
// //after disconnection from server, remove HOSTNAME key from HashMap
.doAfterTerminate(() -> serverConnectivityList.remove(hostname));
} else {
return Single.error(new IllegalStateException("not prepared"));
}
});
} }
serverConnectivityList.put(hostname, ServerConnectivity.STATE_CONNECTED);
connectivitySubject.onNext(
new ServerConnectivity(hostname, ServerConnectivity.STATE_CONNECTED));
}
@DebugLog
@Override
public void notifyConnectionLost(String hostname, int code) {
serverConnectivityList.put(hostname, ServerConnectivity.STATE_DISCONNECTED);
connectivitySubject.onNext(
new ServerConnectivity(hostname, ServerConnectivity.STATE_DISCONNECTED, code));
}
@DebugLog
@Override
public void notifyConnecting(String hostname) {
serverConnectivityList.put(hostname, ServerConnectivity.STATE_CONNECTING);
connectivitySubject.onNext(
new ServerConnectivity(hostname, ServerConnectivity.STATE_CONNECTING));
}
@Override
public Flowable<ServerConnectivity> getServerConnectivityAsObservable() {
return connectivitySubject.toFlowable(BackpressureStrategy.LATEST);
}
@Override
public int getConnectivityState(@NonNull String hostname) {
return serverConnectivityList.get(hostname);
}
@DebugLog
private Single<Boolean> connectToServerIfNeeded(String hostname, boolean forceConnect) {
return Single.defer(() -> {
Integer state = serverConnectivityList.get(hostname);
if (state == null) {
state = ServerConnectivity.STATE_DISCONNECTED;
}
final int connectivity = state;
if (!forceConnect && connectivity == ServerConnectivity.STATE_CONNECTED) {
return Single.just(true);
}
if (connectivity == ServerConnectivity.STATE_DISCONNECTING) {
return waitForDisconnected(hostname)
.flatMap(_val -> connectToServerIfNeeded(hostname, forceConnect));
}
if (connectivity == ServerConnectivity.STATE_DISCONNECTED) {
// notifyConnecting(hostname);
}
return connectToServer(hostname);
});
}
private Single<Boolean> disconnectFromServerIfNeeded(String hostname) {
return Single.defer(() -> {
final int connectivity = serverConnectivityList.get(hostname);
if (connectivity == ServerConnectivity.STATE_DISCONNECTED) {
return Single.just(true);
}
if (connectivity == ServerConnectivity.STATE_CONNECTING) {
return waitForConnected(hostname)
.doOnError(err -> notifyConnectionLost(hostname, DDPClient.REASON_NETWORK_ERROR))
.flatMap(_val -> disconnectFromServerIfNeeded(hostname));
}
if (connectivity == ServerConnectivity.STATE_DISCONNECTING) {
return waitForDisconnected(hostname);
}
return disconnectFromServer(hostname)
.retryWhen(RxHelper.exponentialBackoff(1, 500, TimeUnit.MILLISECONDS));
});
}
@DebugLog
private Single<Boolean> waitForConnected(String hostname) {
return connectivitySubject
.filter(serverConnectivity -> hostname.equals(serverConnectivity.hostname))
.map(serverConnectivity -> serverConnectivity.state)
.filter(state ->
state == ServerConnectivity.STATE_CONNECTED
|| state == ServerConnectivity.STATE_DISCONNECTED)
.firstElement()
.toSingle()
.flatMap(state ->
state == ServerConnectivity.STATE_CONNECTED
? Single.just(true)
: Single.error(new ServerConnectivity.DisconnectedException()));
}
@DebugLog
private Single<Boolean> waitForDisconnected(String hostname) {
return connectivitySubject
.filter(serverConnectivity -> hostname.equals(serverConnectivity.hostname))
.map(serverConnectivity -> serverConnectivity.state)
.filter(state -> state == ServerConnectivity.STATE_DISCONNECTED)
.firstElement()
.toSingle()
.map(state -> true);
}
@DebugLog
private Single<Boolean> connectToServer(String hostname) {
return Single.defer(() -> {
if (!serverConnectivityList.containsKey(hostname)) {
return Single.error(new IllegalArgumentException("hostname not found"));
}
if (serverConnectivityList.get(hostname) != ServerConnectivity.STATE_CONNECTED) {
// Mark as CONNECTING except for the case [forceConnect && connected] because
// ensureConnectionToServer doesn't notify ConnectionEstablished/Lost is already connected.
// serverConnectivityList.put(hostname, ServerConnectivity.STATE_CONNECTING);
}
if (serviceInterface != null) {
return serviceInterface.ensureConnectionToServer(hostname);
} else {
return Single.error(new IllegalStateException("not prepared"));
}
});
}
private Single<Boolean> disconnectFromServer(String hostname) {
return Single.defer(() -> {
if (!serverConnectivityList.containsKey(hostname)) {
return Single.error(new IllegalArgumentException("hostname not found"));
}
serverConnectivityList.put(hostname, ServerConnectivity.STATE_DISCONNECTING);
if (serviceInterface != null) {
return serviceInterface.disconnectFromServer(hostname)
// //after disconnection from server, remove HOSTNAME key from HashMap
.doAfterTerminate(() -> serverConnectivityList.remove(hostname));
} else {
return Single.error(new IllegalStateException("not prepared"));
}
});
}
} }
\ No newline at end of file
app/src/main/java/chat/rocket/android/service/RocketChatService.java
View file @ 87461e2a
...@@ -23,130 +23,130 @@ import io.reactivex.Single; ...@@ -23,130 +23,130 @@ import io.reactivex.Single;
*/ */
public class RocketChatService extends Service implements ConnectivityServiceInterface { public class RocketChatService extends Service implements ConnectivityServiceInterface {
private ConnectivityManagerInternal connectivityManager; private ConnectivityManagerInternal connectivityManager;
private static volatile Semaphore webSocketThreadLock = new Semaphore(1); private static volatile Semaphore webSocketThreadLock = new Semaphore(1);
private static volatile RocketChatWebSocketThread currentWebSocketThread; private static volatile RocketChatWebSocketThread currentWebSocketThread;
public class LocalBinder extends Binder {
ConnectivityServiceInterface getServiceInterface() {
return RocketChatService.this;
}
}
private final LocalBinder localBinder = new LocalBinder();
/**
* ensure RocketChatService alive.
*/
/*package*/static void keepAlive(Context context) {
context.startService(new Intent(context, RocketChatService.class));
}
public static void bind(Context context, ServiceConnection serviceConnection) {
context.bindService(
new Intent(context, RocketChatService.class), serviceConnection, Context.BIND_AUTO_CREATE);
}
public static void unbind(Context context, ServiceConnection serviceConnection) {
context.unbindService(serviceConnection);
}
@DebugLog
@Override
public void onCreate() {
super.onCreate();
connectivityManager = ConnectivityManager.getInstanceForInternal(getApplicationContext());
connectivityManager.resetConnectivityStateList();
}
public class LocalBinder extends Binder { @DebugLog
ConnectivityServiceInterface getServiceInterface() { @Override
return RocketChatService.this; public int onStartCommand(Intent intent, int flags, int startId) {
connectivityManager.ensureConnections();
return START_NOT_STICKY;
} }
}
@Override
private final LocalBinder localBinder = new LocalBinder(); public Single<Boolean> ensureConnectionToServer(String hostname) { //called via binder.
return getOrCreateWebSocketThread(hostname)
/** .flatMap(RocketChatWebSocketThread::keepAlive);
* ensure RocketChatService alive. }
*/
/*package*/ static void keepAlive(Context context) { @Override
context.startService(new Intent(context, RocketChatService.class)); public Single<Boolean> disconnectFromServer(String hostname) { //called via binder.
} return Single.defer(() -> {
if (!existsThreadForHostname(hostname)) {
public static void bind(Context context, ServiceConnection serviceConnection) { return Single.just(true);
context.bindService( }
new Intent(context, RocketChatService.class), serviceConnection, Context.BIND_AUTO_CREATE);
} if (currentWebSocketThread != null) {
return currentWebSocketThread.terminate()
public static void unbind(Context context, ServiceConnection serviceConnection) { // after disconnection from server
context.unbindService(serviceConnection); .doAfterTerminate(() -> {
}
@DebugLog
@Override
public void onCreate() {
super.onCreate();
connectivityManager = ConnectivityManager.getInstanceForInternal(getApplicationContext());
connectivityManager.resetConnectivityStateList();
}
@DebugLog
@Override
public int onStartCommand(Intent intent, int flags, int startId) {
connectivityManager.ensureConnections();
return START_NOT_STICKY;
}
@Override
public Single<Boolean> ensureConnectionToServer(String hostname) { //called via binder.
return getOrCreateWebSocketThread(hostname)
.flatMap(RocketChatWebSocketThread::keepAlive);
}
@Override
public Single<Boolean> disconnectFromServer(String hostname) { //called via binder.
return Single.defer(() -> {
if (!existsThreadForHostname(hostname)) {
return Single.just(true);
}
if (currentWebSocketThread != null) {
return currentWebSocketThread.terminate()
// after disconnection from server
.doAfterTerminate(() -> {
currentWebSocketThread = null;
// remove RealmConfiguration key from HashMap
RealmStore.sStore.remove(hostname);
});
} else {
return Observable.timer(1, TimeUnit.SECONDS).singleOrError()
.flatMap(_val -> disconnectFromServer(hostname));
}
});
}
@DebugLog
private Single<RocketChatWebSocketThread> getOrCreateWebSocketThread(String hostname) {
return Single.defer(() -> {
webSocketThreadLock.acquire();
int connectivityState = ConnectivityManager.getInstance(getApplicationContext()).getConnectivityState(hostname);
boolean isDisconnected = connectivityState != ServerConnectivity.STATE_CONNECTED;
if (currentWebSocketThread != null && existsThreadForHostname(hostname) && !isDisconnected) {
webSocketThreadLock.release();
return Single.just(currentWebSocketThread);
}
if (currentWebSocketThread != null) {
return currentWebSocketThread.terminate()
.doAfterTerminate(() -> currentWebSocketThread = null)
.flatMap(terminated ->
RocketChatWebSocketThread.getStarted(getApplicationContext(), hostname)
.doOnSuccess(thread -> {
currentWebSocketThread = thread;
webSocketThreadLock.release();
})
.doOnError(throwable -> {
currentWebSocketThread = null; currentWebSocketThread = null;
RCLog.e(throwable); // remove RealmConfiguration key from HashMap
Logger.report(throwable); RealmStore.sStore.remove(hostname);
webSocketThreadLock.release(); });
}) } else {
); return Observable.timer(1, TimeUnit.SECONDS).singleOrError()
} .flatMap(_val -> disconnectFromServer(hostname));
}
return RocketChatWebSocketThread.getStarted(getApplicationContext(), hostname) });
.doOnSuccess(thread -> { }
currentWebSocketThread = thread;
webSocketThreadLock.release(); @DebugLog
}) private Single<RocketChatWebSocketThread> getOrCreateWebSocketThread(String hostname) {
.doOnError(throwable -> { return Single.defer(() -> {
currentWebSocketThread = null; webSocketThreadLock.acquire();
RCLog.e(throwable); int connectivityState = ConnectivityManager.getInstance(getApplicationContext()).getConnectivityState(hostname);
Logger.report(throwable); boolean isDisconnected = connectivityState != ServerConnectivity.STATE_CONNECTED;
webSocketThreadLock.release(); if (currentWebSocketThread != null && existsThreadForHostname(hostname) && !isDisconnected) {
}); webSocketThreadLock.release();
}); return Single.just(currentWebSocketThread);
} }
private boolean existsThreadForHostname(String hostname) { if (currentWebSocketThread != null) {
if (hostname == null || currentWebSocketThread == null) { return currentWebSocketThread.terminate()
return false; .doAfterTerminate(() -> currentWebSocketThread = null)
.flatMap(terminated ->
RocketChatWebSocketThread.getStarted(getApplicationContext(), hostname)
.doOnSuccess(thread -> {
currentWebSocketThread = thread;
webSocketThreadLock.release();
})
.doOnError(throwable -> {
currentWebSocketThread = null;
RCLog.e(throwable);
Logger.report(throwable);
webSocketThreadLock.release();
})
);
}
return RocketChatWebSocketThread.getStarted(getApplicationContext(), hostname)
.doOnSuccess(thread -> {
currentWebSocketThread = thread;
webSocketThreadLock.release();
})
.doOnError(throwable -> {
currentWebSocketThread = null;
RCLog.e(throwable);
Logger.report(throwable);
webSocketThreadLock.release();
});
});
}
private boolean existsThreadForHostname(String hostname) {
if (hostname == null || currentWebSocketThread == null) {
return false;
}
return currentWebSocketThread.getName().equals("RC_thread_" + hostname);
}
@Nullable
@Override
public IBinder onBind(Intent intent) {
return localBinder;
} }
return currentWebSocketThread.getName().equals("RC_thread_" + hostname);
}
@Nullable
@Override
public IBinder onBind(Intent intent) {
return localBinder;
}
} }
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