Mercury-IM/domain/src/main/java/org/mercury_im/messenger/xmpp/MercuryConnection.java

package org.mercury_im.messenger.xmpp;

import org.jivesoftware.smack.AbstractXMPPConnection;
import org.jivesoftware.smack.ConnectionListener;
import org.jivesoftware.smack.SmackException;
import org.jivesoftware.smack.XMPPConnection;
import org.jivesoftware.smack.XMPPException;
import org.mercury_im.messenger.data.repository.AccountRepository;
import org.mercury_im.messenger.entity.Account;
import org.mercury_im.messenger.util.Optional;

import java.io.IOException;
import java.util.logging.Level;
import java.util.logging.Logger;

import io.reactivex.Observable;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.disposables.Disposable;
import io.reactivex.subjects.BehaviorSubject;

public class MercuryConnection {

    private static final Logger LOGGER = Logger.getLogger("MercuryConnection");

    private Account account;
    private XMPPConnection connection;

    private BehaviorSubject<Boolean> enabled = BehaviorSubject.create();
    private BehaviorSubject<ConnectionState> state = BehaviorSubject.createDefault(ConnectionState.closed);

    public MercuryConnection(Account account, XMPPConnection connection) {
        this.account = account;
        this.connection = connection;
    }

    public Observable<ConnectionState> getState() {
        return state;
    }

    public void observeAccountAndConnection(AccountRepository accountRepository) {
        observeAccount(accountRepository);
        observeConnection();
    }

    private Disposable observeAccount(AccountRepository accountRepository) {
        return accountRepository.observeAccount(account.getId())
                .filter(Optional::isPresent)
                .map(Optional::getItem)
                .subscribe(this::setAccount);
    }

    private void observeConnection() {
        // TODO: Remove listener at some point
        connection.addConnectionListener(connectionListener);
    }

    private void setAccount(Account account) {
        this.account = account;
        enabled.onNext(account.isEnabled());
    }

    public Account getAccount() {
        return account;
    }

    public XMPPConnection getConnection() {
        return connection;
    }

    private final ConnectionListener connectionListener = new ConnectionListener() {
        @Override
        public void connected(XMPPConnection connection) {
            LOGGER.log(Level.FINER, "connected");
            state.onNext(ConnectionState.connected);
        }

        @Override
        public void authenticated(XMPPConnection connection, boolean resumed) {
            LOGGER.log(Level.FINER, "authenticated. resumed? " + resumed);
            state.onNext(ConnectionState.authenticated);
        }

        @Override
        public void connectionClosed() {
            LOGGER.log(Level.FINER, "connectionClosed");
            state.onNext(ConnectionState.closed);
        }

        @Override
        public void connectionClosedOnError(Exception e) {
            LOGGER.log(Level.WARNING, "connectionClosedOnError");
            state.onNext(ConnectionState.closedOnError);
        }
    };

    @Override
    public String toString() {
        return "MercuryConnection(" + getAccount() + ")";
    }

    public void login() throws InterruptedException, XMPPException, SmackException, IOException {
        ((AbstractXMPPConnection) connection).connect().login();
    }

    public enum ConnectionState {
        connected,
        authenticated,
        closedOnError,
        closed
    }
}