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.entity.Account; import org.mercury_im.messenger.usecase.LogIntoAccount; 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 io.reactivex.Observable; import io.reactivex.disposables.CompositeDisposable; import io.reactivex.subjects.BehaviorSubject; 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 Messenger messenger; private final CompositeDisposable disposable = new CompositeDisposable(); @Inject public MercuryConnectionManager(Messenger messenger, AccountRepository accountRepository) { this.messenger = messenger; this.accountRepository = accountRepository; } 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 registerConnection(MercuryConnection connection) { putConnection(connection); if (connection.getConnection().isAuthenticated()) { registerLiveConnection(connection); } else { registerDeadConnection(connection); } } private void putConnection(MercuryConnection connection) { connections.put(connection.getAccount().getId(), connection); connectionsSubject.onNext(connections); messenger.bindConnection(connection); } private void registerLiveConnection(MercuryConnection connection) { Observable> observableAccount = accountRepository .observeAccount(connection.getAccount().getId()); disposable.add(observableAccount.subscribe(event -> handleOptionalAccountChangedEvent(connection, event))); } private void registerDeadConnection(MercuryConnection connection) { if (connection.getAccount().isEnabled()) { connectionLogin(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) { shutdownConnection(connection); } private void handleAccountEnabled(MercuryConnection connection) { connectionLogin(connection); } private void connectionLogin(MercuryConnection connection) { disposable.add(LogIntoAccount.with(connection).executeAndPossiblyThrow() .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()).instantShutdown(); } } private void removeConnection(MercuryConnection connection) { connections.remove(connection.getAccount().getId()); connectionsSubject.onNext(connections); connection.dispose(); } }