Mercury-IM/core/src/main/java/org/mercury_im/messenger/core/ConnectionCenter.java

package org.mercury_im.messenger.core;

import org.jivesoftware.smack.AbstractXMPPConnection;
import org.jivesoftware.smack.tcp.XMPPTCPConnection;
import org.jivesoftware.smack.tcp.XMPPTCPConnectionConfiguration;
import org.jivesoftware.smackx.caps.EntityCapsManager;
import org.mercury_im.messenger.persistence.model.AccountModel;
import org.mercury_im.messenger.persistence.repository.AccountRepository;

import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.inject.Inject;
import javax.inject.Singleton;

import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;

@Singleton
public class ConnectionCenter {

    // Injected
    private final AccountRepository accountRepository;
    private final EntityCapsStore entityCapsStore;

    // Connections
    private final Map<Long, MercuryConnection> connectionMap =
            Collections.synchronizedMap(new HashMap<>());

    // Disposable for rx
    private final CompositeDisposable disposable = new CompositeDisposable();

    private final AtomicBoolean started = new AtomicBoolean(false);

    @Inject
    public ConnectionCenter(EntityCapsStore capsStore, AccountRepository accountRepository) {
        this.accountRepository = accountRepository;
        this.entityCapsStore = capsStore;

        EntityCapsManager.setPersistentCache(capsStore);
    }

    /**
     * Start up the center by subscribing to changes of the {@link AccountModel accounts} in the
     * database. For each new {@link AccountModel} it creates a {@link MercuryConnection} and
     * stores it in the {@link #connectionMap}.
     */
    @SuppressWarnings("unchecked")
    public synchronized void startUp() {

        if (started.getAndSet(true)) {
            // already started.
            return;
        }

        // otherwise subscribe to accounts and create connections.
        disposable.add(accountRepository.getAllAccounts()
                .observeOn(Schedulers.io())
                .subscribeOn(Schedulers.computation())
                .subscribe((Consumer<List<? extends AccountModel>>) accounts -> {
                    for (AccountModel account : accounts) {
                        if (connectionMap.get(account.getId()) != null) {
                            continue;
                        }
                        MercuryConnection connection = createConnection(account);
                        connectionMap.put(account.getId(), connection);
                    }
                }));
    }

    public MercuryConnection getConnection(AccountModel account) {
        return getConnection(account.getId());
    }

    public MercuryConnection getConnection(long accountId) {
        return connectionMap.get(accountId);
    }

    public void putConnection(MercuryConnection connection) {
        connectionMap.put(connection.getAccountId(), connection);
    }

    public MercuryConnection createConnection(AccountModel accountModel) {
        XMPPTCPConnectionConfiguration configuration = XMPPTCPConnectionConfiguration.builder()
                .setHost(accountModel.getJid().getDomain().toString())
                .setXmppAddressAndPassword(accountModel.getJid(), accountModel.getPassword())
                .setConnectTimeout(2 * 60 * 1000)
                .setEnabledSSLCiphers(MercuryConfiguration.enabledCiphers)
                .setEnabledSSLProtocols(MercuryConfiguration.enabledProtocols)
                .build();

        AbstractXMPPConnection tcpConnection = new XMPPTCPConnection(configuration);
        return new MercuryConnection(tcpConnection, accountModel.getId());
    }
}