diff --git a/core/src/main/java/org/mercury_im/messenger/core/centers/ConnectionCenter.java b/core/src/main/java/org/mercury_im/messenger/core/centers/ConnectionCenter.java index 3e19f83..2f3e6e8 100644 --- a/core/src/main/java/org/mercury_im/messenger/core/centers/ConnectionCenter.java +++ b/core/src/main/java/org/mercury_im/messenger/core/centers/ConnectionCenter.java @@ -30,6 +30,7 @@ import javax.inject.Inject; import javax.inject.Singleton; import io.reactivex.disposables.CompositeDisposable; +import io.reactivex.disposables.Disposable; import io.reactivex.functions.Consumer; import io.reactivex.schedulers.Schedulers; @@ -82,50 +83,54 @@ public class ConnectionCenter { } // otherwise subscribe to accounts and create connections. - disposable.add( - accountRepository.getAllAccounts() - .observeOn(Schedulers.io()) - .subscribeOn(Schedulers.computation()) - .subscribe((Consumer>) accounts -> { - LOGGER.log(Level.INFO, "Accounts changed."); - Set accountIds = new HashSet<>(); - // Add missing connections to the map - for (AccountModel account : accounts) { - accountIds.add(account.getId()); - if (connectionMap.get(account.getId()) != null) { - continue; - } + Disposable allAccounts = accountRepository.getAllAccounts() + .observeOn(Schedulers.io()) + .subscribeOn(Schedulers.computation()) + .subscribe((Consumer>) accounts -> { + LOGGER.log(Level.INFO, "Accounts changed."); + Set accountIds = new HashSet<>(); - LOGGER.log(Level.INFO, "Add new connection " + account.getJid().toString() + " to ConnectionCenter list."); - MercuryConnection connection = createConnection(account); - connectionMap.put(account.getId(), connection); + // Add missing connections to the map + for (AccountModel account : accounts) { + accountIds.add(account.getId()); + if (connectionMap.get(account.getId()) != null) { + continue; + } - RosterStore rosterStore = new RosterStore(rosterRepository); - rosterStore.setAccountId(account.getId()); - rosterStore.subscribe(); - connection.getRoster().setRosterStore(rosterStore); + LOGGER.log(Level.INFO, "Add new connection " + account.getJid().toString() + " to ConnectionCenter list."); + MercuryConnection connection = createConnection(account); + connectionMap.put(account.getId(), connection); - messageStore.registerForMercuryConnection(connection); + // initialize new connection + RosterStore rosterStore = new RosterStore(rosterRepository); + rosterStore.setAccountId(account.getId()); + rosterStore.subscribe(); + connection.getRoster().setRosterStore(rosterStore); - if (account.getEnabled()) { - LOGGER.log(Level.INFO, "Connecting..."); - connection.connect(); - LOGGER.log(Level.INFO, "Connected!"); - } else { - LOGGER.log(Level.INFO, "Account not enabled. Skip."); - } - } + messageStore.registerForMercuryConnection(connection); + + if (account.getEnabled()) { + LOGGER.log(Level.INFO, "Connecting..."); + connection.connect(); + LOGGER.log(Level.INFO, "Connected!"); + } else { + LOGGER.log(Level.INFO, "Account not enabled. Skip."); + } + } + + // Remove unwanted connections from the map + for (long connectionId : connectionMap.keySet()) { + if (!accountIds.contains(connectionId)) { + AbstractXMPPConnection con = + (AbstractXMPPConnection) connectionMap.get(connectionId).getConnection(); + con.disconnect(); + connectionMap.remove(connectionId); + } + } + }); + + disposable.add(allAccounts); - // Remove unwanted connections from the map - for (long connectionId : connectionMap.keySet()) { - if (!accountIds.contains(connectionId)) { - AbstractXMPPConnection con = - (AbstractXMPPConnection) connectionMap.get(connectionId).getConnection(); - con.disconnect(); - connectionMap.remove(connectionId); - } - } - })); } public MercuryConnection getConnection(AccountModel account) { @@ -140,6 +145,15 @@ public class ConnectionCenter { connectionMap.put(connection.getAccountId(), connection); } + /** + * Create a new {@link MercuryConnection} with an underlying {@link XMPPTCPConnection} + * from the credentials of an {@link AccountModel}. + * The new connection will not be connected. + * + * @param accountModel accountModel + * + * @return disconnected mercury connection + */ public MercuryConnection createConnection(AccountModel accountModel) { LOGGER.log(Level.INFO, "Create Connection for " + accountModel.getJid().toString()); XMPPTCPConnectionConfiguration configuration = XMPPTCPConnectionConfiguration.builder() @@ -154,6 +168,11 @@ public class ConnectionCenter { return new MercuryConnection(tcpConnection, accountModel.getId()); } + /** + * Set Client State Indication status to active. + * + * @see XEP-0352: Client State Indication + */ public void clientStateActive() { LOGGER.log(Level.INFO, "CSI: App is going to foreground -> active"); for (MercuryConnection mercuryConnection : connectionMap.values()) { @@ -161,15 +180,18 @@ public class ConnectionCenter { if (connection.isConnected() && ClientStateIndicationManager.isSupported(connection)) { try { ClientStateIndicationManager.active(mercuryConnection.getConnection()); - } catch (SmackException.NotConnectedException e) { - e.printStackTrace(); - } catch (InterruptedException e) { + } catch (SmackException.NotConnectedException | InterruptedException e) { e.printStackTrace(); } } } } + /** + * Set Client State Indication status to inactive. + * + * @see XEP-0352: Client State Indication + */ public void clientStateInactive() { LOGGER.log(Level.INFO, "CSI: App is going to background -> inactive"); for (MercuryConnection mercuryConnection : connectionMap.values()) { @@ -177,9 +199,7 @@ public class ConnectionCenter { if (connection.isConnected() && ClientStateIndicationManager.isSupported(connection)) { try { ClientStateIndicationManager.inactive(connection); - } catch (SmackException.NotConnectedException e) { - e.printStackTrace(); - } catch (InterruptedException e) { + } catch (SmackException.NotConnectedException | InterruptedException e) { e.printStackTrace(); } }