Class that containts all socket logic.

SocketManagerImpl

package com.twelve.app.data.network.repositories.impl;

import android.annotation.SuppressLint;
import android.util.Log;

import com.google.gson.internal.LinkedTreeMap;
import com.google.gson.reflect.TypeToken;
import com.microsoft.signalr.HubConnection;
import com.microsoft.signalr.HubConnectionBuilder;
import com.microsoft.signalr.HubConnectionState;
import com.twelve.app.data.Constants;
import com.twelve.app.data.models.common.Story;
import com.twelve.app.data.models.response.MatchXgResponse;
import com.twelve.app.data.models.socket.BaseSocketEvent;
import com.twelve.app.data.models.socket.events.AlreadyConnectedEvent;
import com.twelve.app.data.models.socket.events.ConnectionSuccessEvent;
import com.twelve.app.data.models.socket.events.FavPlayerPointsEvent;
import com.twelve.app.data.models.socket.events.MatchEndedEvent;
import com.twelve.app.data.models.socket.events.MatchGoalsEvent;
import com.twelve.app.data.models.socket.events.MatchStartedEvent;
import com.twelve.app.data.models.socket.events.MatchStoriesEvent;
import com.twelve.app.data.models.socket.events.MatchSummaryEvent;
import com.twelve.app.data.models.socket.pojo.SummaryResponse;
import com.twelve.app.data.network.repositories.SocketManager;
import com.twelve.app.data.network.repositories.UserAuthHelper;

import java.util.List;

import javax.inject.Inject;

import io.reactivex.Completable;
import io.reactivex.Observable;
import io.reactivex.Single;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;

public class SocketManagerImpl implements SocketManager {

    public final static String TAG = "test_socket";
    public final static int CONNECTION_TIMEOUT = 40_000; //milliseconds

    //events that are sent.
    public final static String SOCKET_EVENT_SELECT_PLAYER = "SelectPlayer";
    public final static String SOCKET_EVENT_DESELECT_PLAYER = "DeselectPlayer";

    //events that are received.
    public final static String SOCKET_EVENT_PLAYER_POINTS = "PlayerPoint";
    public final static String SOCKET_EVENT_START_MATCH = "MatchStart";
    public final static String SOCKET_EVENT_END_MATCH = "MatchEnd";
    public final static String SOCKET_EVENT_MATCH_SUMMARY = "MatchSummary";
    public final static String SOCKET_EVENT_MATCH_STORIES = "MatchStories";
    public final static String SOCKET_EVENT_MATCH_GOALS = "MatchGoals";

    private final UserAuthHelper mUserAuthHelper;
    private HubConnection mHubConnection;
    private Disposable mStartConnectionDisposable;
    private long mFavPlayerId = -1;
    private long mMatchId = -1;

    @Inject
    public SocketManagerImpl(UserAuthHelper userAuthHelper) {
        mUserAuthHelper = userAuthHelper;
        init();
    }

    private void init() {
        mHubConnection = HubConnectionBuilder.create(Constants.Network.SOCKET_URL)
                .withAccessTokenProvider(Single.fromCallable(() -> {
                    Log.d(SocketManagerImpl.TAG, "Token provider inside SignalR..");
                    // If token is expired then need to wait until token will be updated:
                    while (mUserAuthHelper.getTokenExpiredTime() <= System.currentTimeMillis()) {
                        // Looped cycle;
                        Log.d(SocketManagerImpl.TAG, "loop... Thread: " + Thread.currentThread().getName());
                    }
                    return mUserAuthHelper.getToken();
                }))
                .build();
        mHubConnection.setServerTimeout(CONNECTION_TIMEOUT);
    }

    @SuppressLint("LogNotTimber")
    @Override
    public Observable<BaseSocketEvent> connect(long matchId, long homeTeamId) {
        mFavPlayerId = -1;
        mMatchId = matchId;
        return Observable.create(emitter -> {
            //checks
            if (emitter.isDisposed()) {
                emitter.onError(new Exception("SocketManagerImpl.connect(). Start connection disposable is disposed!"));
                return;
            }

            if (mHubConnection.getConnectionState() == HubConnectionState.CONNECTED) {
                emitter.onNext(new AlreadyConnectedEvent());
                return;
            }

            //set url
            mHubConnection.setBaseUrl(Constants.Network.SOCKET_URL + mMatchId);

            //Events:
            //new points of  favorite player;
            mHubConnection.on(SOCKET_EVENT_PLAYER_POINTS, message -> {
                emitter.onNext(new FavPlayerPointsEvent(message));
            }, LinkedTreeMap.class);

            //match started
            mHubConnection.on(SOCKET_EVENT_START_MATCH, message -> {
                emitter.onNext(new MatchStartedEvent());
            }, Object.class);

            //match ended
            mHubConnection.on(SOCKET_EVENT_END_MATCH, message -> {
                emitter.onNext(new MatchEndedEvent());
            }, Object.class);

            //match summary
            mHubConnection.on(SOCKET_EVENT_MATCH_SUMMARY, message -> {
                emitter.onNext(new MatchSummaryEvent((List<SummaryResponse>) message, homeTeamId));
            }, new TypeToken<List<SummaryResponse>>() {
            }.getType());

            //match stories
            mHubConnection.on(SOCKET_EVENT_MATCH_STORIES, message -> {
                emitter.onNext(new MatchStoriesEvent((List<Story>) message));
            }, new TypeToken<List<Story>>() {
            }.getType());

            //match goals
            mHubConnection.on(SOCKET_EVENT_MATCH_GOALS, message -> {
                emitter.onNext(new MatchGoalsEvent(message));
            }, MatchXgResponse.class);

            //open connection
            mStartConnectionDisposable = startConnection()
                    .subscribe(() -> emitter.onNext(new ConnectionSuccessEvent()),
                            emitter::onError);
        })
                .map(object -> (BaseSocketEvent) object)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread());
    }

    @SuppressLint("LogNotTimber")
    @Override
    public Completable disconnect() {
        mFavPlayerId = -1;
        mMatchId = -1;
        Log.d(TAG, "disconnect method. State: " + mHubConnection.getConnectionState().toString());
        if (mHubConnection.getConnectionState() == HubConnectionState.DISCONNECTED)
            return Completable.error(new Throwable("Not Connected. No need to disconnect."));

        if (mStartConnectionDisposable != null && !mStartConnectionDisposable.isDisposed()) {
            Log.d(TAG, "SocketManagerImpl.disconnect(). mStartConnectionDisposable.isDisposed() = " + mStartConnectionDisposable.isDisposed());
            mStartConnectionDisposable.dispose();
        }

        if (mHubConnection == null)
            return Completable.error(new Throwable("mHubConnection is null. No need to stop connection."));

        // WORKAROUND! crash inside library, that's why the following code is wrapped in try\catch block;
        try {
            return mHubConnection.stop();
        } catch (Exception e) {
            Log.d(TAG, "CRASH: " + e.getMessage());
            return Completable.error(e);
        }
    }

    private Completable startConnection() {
        return mHubConnection.start();
    }

    @Override
    public Completable selectPlayer(long matchId, long playerId) {
        if (mHubConnection.getConnectionState() != HubConnectionState.CONNECTED)
            return Completable.error(new Throwable("Selecting player has not been made. No connection."));

        if (playerId == -1)
            return Completable.error(new Throwable("Selecting player has not been made. No favorite player selected."));

        if (mFavPlayerId == playerId)
            return Completable.error(new Throwable("No need to select the same player."));

        long oldPlayerId = mFavPlayerId;
        mFavPlayerId = playerId;
        if (oldPlayerId != -1) {
            return sendDeselectPlayerEvent(matchId, oldPlayerId)
                    .andThen(Completable.defer(() -> sendSelectPlayerEvent(matchId, mFavPlayerId)));
        } else {
            return sendSelectPlayerEvent(matchId, mFavPlayerId);
        }
    }

    private Completable sendSelectPlayerEvent(long matchId, long playerId) {
        return mHubConnection.invoke(SOCKET_EVENT_SELECT_PLAYER, matchId, playerId)
                .doOnComplete(() -> Log.d(TAG, "Selected player with id " + playerId));
    }

    private Completable sendDeselectPlayerEvent(long matchId, long playerId) {
        return mHubConnection.invoke(SOCKET_EVENT_DESELECT_PLAYER, matchId, playerId)
                .doOnComplete(() -> Log.d(TAG, "Deselected player with id " + playerId));
    }
}