package org.mercury_im.messenger.xmpp; import org.jivesoftware.smack.AbstractXMPPConnection; import org.mercury_im.messenger.Messenger; import org.mercury_im.messenger.data.repository.AccountRepository; import org.mercury_im.messenger.data.repository.Repositories; import org.mercury_im.messenger.entity.Account; import org.mercury_im.messenger.usecase.LogIntoAccount; import org.mercury_im.messenger.usecase.RosterStoreBinder; import org.mercury_im.messenger.util.Optional; import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.logging.Level; import java.util.logging.Logger; import javax.inject.Inject; import javax.inject.Singleton; import io.reactivex.Observable; import io.reactivex.Scheduler; import io.reactivex.disposables.CompositeDisposable; import io.reactivex.schedulers.Schedulers; import io.reactivex.subjects.BehaviorSubject; @Singleton public class MercuryConnectionManager { private static final Logger LOGGER = Logger.getLogger(MercuryConnectionManager.class.getName()); private final Map connections = new ConcurrentHashMap<>(); private final BehaviorSubject> connectionsSubject = BehaviorSubject.createDefault(connections); private final AccountRepository accountRepository; private final RosterStoreBinder rosterStoreBinder; private final CompositeDisposable disposable = new CompositeDisposable(); @Inject public MercuryConnectionManager(Repositories repositories, RosterStoreBinder rosterStoreBinder) { this.accountRepository = repositories.getAccountRepository(); this.rosterStoreBinder = rosterStoreBinder; } public List getConnections() { return new ArrayList<>(connections.values()); } public Observable> observeConnections() { return connectionsSubject; } public MercuryConnection getConnection(Account account) { return getConnection(account.getId()); } public MercuryConnection getConnection(UUID id) { return connections.get(id); } public void registerConnections(List accounts) { for (Account account : accounts) { MercuryConnection connection = new MercuryConnection(accountRepository, account); registerConnection(connection); } } public void registerConnection(MercuryConnection connection) { LOGGER.log(Level.INFO, "Register Connection " + connection.getAccount().getAddress()); putConnection(connection); disposable.add(accountRepository .observeAccount(connection.getAccount().getId()) .subscribeOn(Schedulers.newThread()) .observeOn(Schedulers.newThread()) .subscribe(event -> handleOptionalAccountChangedEvent(connection, event))); } private void putConnection(MercuryConnection connection) { connections.put(connection.getAccount().getId(), connection); connectionsSubject.onNext(connections); bindConnection(connection); } public void bindConnection(MercuryConnection connection) { rosterStoreBinder.setRosterStoreOn(connection); } private void handleOptionalAccountChangedEvent(MercuryConnection connection, Optional event) { if (event.isPresent()) { handleAccountChangedEvent(connection, event.getItem()); } else { handleAccountRemoved(connection); } } private void handleAccountChangedEvent(MercuryConnection connection, Account account) { if (account.isEnabled()) { handleAccountEnabled(connection); } else { handleAccountDisabled(connection); } } private void handleAccountDisabled(MercuryConnection connection) { LOGGER.log(Level.INFO, "HandleAccountDisabled: " + connection.getAccount().getAddress()); shutdownConnection(connection); } private void handleAccountEnabled(MercuryConnection connection) { LOGGER.log(Level.INFO, "HandleAccountEnabled: " + connection.getAccount().getAddress()); connectionLogin(connection); } private void connectionLogin(MercuryConnection connection) { disposable.add(LogIntoAccount.with(connection).executeAndPossiblyThrow() .subscribeOn(Schedulers.newThread()) .subscribe(() -> LOGGER.log(Level.INFO, "Logged in."), error -> LOGGER.log(Level.SEVERE, "Connection error!", error))); } private void handleAccountRemoved(MercuryConnection connection) { disconnectAndRemoveConnection(connection); } private void disconnectAndRemoveConnection(MercuryConnection connection) { shutdownConnection(connection); removeConnection(connection); } private void shutdownConnection(MercuryConnection connection) { if (connection.getConnection().isAuthenticated()) { ((AbstractXMPPConnection) connection.getConnection()).disconnect(); } } private void removeConnection(MercuryConnection connection) { connections.remove(connection.getAccount().getId()); connectionsSubject.onNext(connections); connection.dispose(); } }