Base class to do network requests and children of it to do particular job

RequestAction.kt

package com.wellaapp.domain.action

import android.content.Context
import android.view.View
import com.wellaapp.data.error_handler.RetrofitException
import com.wellaapp.presentation.extensions.isInternetConnectionExists
import io.reactivex.Observable
import io.reactivex.android.schedulers.AndroidSchedulers
import io.reactivex.schedulers.Schedulers

/**
 * @author Denis_Zinkovskiy at 12/29/16.
 *
 */
abstract class RequestAction<M, in T> : BaseAction<M,T>() {

    override fun fireAction(context: Context, view: View?, actionType: String?, model: T): Observable<M> {
        if(context.isInternetConnectionExists()){
            return prepareObservable(model)
                    .subscribeOn(Schedulers.computation())
                    .observeOn(AndroidSchedulers.mainThread())

        }else return Observable.error { RetrofitException.networkError(null,"No Internet connection") }
    }
}

sessions_actions.kt

package com.wellaapp.domain.action.sessions

import com.wellaapp.data.entities.User
import com.wellaapp.data.model.LoginResponseModel
import com.wellaapp.data.model.Token
import com.wellaapp.data.storage.sessions.SessionStorage
import com.wellaapp.data.storage.users.UsersStorage
import com.wellaapp.domain.action.BaseAction
import com.wellaapp.domain.action.RequestAction
import com.wellaapp.presentation.base.BaseDisposableObserver
import com.wellaapp.presentation.base.logout
import com.wellaapp.presentation.quickblox.ChatHelper
import io.reactivex.Observable
import io.reactivex.android.schedulers.AndroidSchedulers
import io.reactivex.schedulers.Schedulers
import io.realm.Realm
import javax.inject.Inject

/**
 * @author Denis_Zinkovskiy at 1/5/17.
 */

class AnonymousLogin @Inject constructor(
        private var sessionCloudStorage: SessionStorage.CloudStorage,
        private var usersStorage: UsersStorage.LocalStorage,
        private var tokenLocalStorage: SessionStorage.TokenLocalStorage) : RequestAction<LoginResponseModel, User?>() {
    override fun prepareObservable(param: User?): Observable<LoginResponseModel> = sessionCloudStorage.anonymous()
            .doOnNext { ChatHelper.login(it.user.toQBUser()).subscribe(BaseDisposableObserver()) }
            .observeOn(AndroidSchedulers.mainThread())
            .doOnNext { usersStorage.write(it.user); tokenLocalStorage.write(Token(it.token)) }
}

class Login @Inject constructor(private var sessionCloudStorage: SessionStorage.CloudStorage,
                                private var usersStorage: UsersStorage.LocalStorage,
                                private var tokenLocalStorage: SessionStorage.TokenLocalStorage) : RequestAction<LoginResponseModel, User>() {
    override fun prepareObservable(param: User): Observable<LoginResponseModel> =
            sessionCloudStorage.login(param)
                    .doOnNext { ChatHelper.login(it.user.toQBUser()).subscribe(BaseDisposableObserver()) }
                    .observeOn(AndroidSchedulers.mainThread())
                    .doOnNext {
                        it.user.password = param.password; usersStorage.write(it.user)
                        tokenLocalStorage.write(Token(it.token))
                    }
}

class LoginFacebook @Inject constructor(private var sessionCloudStorage: SessionStorage.CloudStorage,
                                        private var usersStorage: UsersStorage.LocalStorage,
                                        private var tokenLocalStorage: SessionStorage.TokenLocalStorage) :
        RequestAction<LoginResponseModel, String>() {
    override fun prepareObservable(param: String): Observable<LoginResponseModel> =
            sessionCloudStorage.facebookLogin(param)
                    .doOnNext { ChatHelper.login(it.user.toQBUser())
                            .doOnError { e -> e.printStackTrace() }
                            .subscribe(BaseDisposableObserver()) }
                    .observeOn(AndroidSchedulers.mainThread())
                    .doOnNext {
                        it.user.isFacebookAuthorized = true; usersStorage.write(it.user)
                        tokenLocalStorage.write(Token(it.token))
                    }
}

class Logout @Inject constructor(private val sessionCloudStorage: SessionStorage.CloudStorage,
                                 private val database: Realm) : RequestAction<Unit?, Token>() {
    override fun prepareObservable(param: Token): Observable<Unit?> =
            sessionCloudStorage.logout(param)
                    .subscribeOn(Schedulers.computation())
                    .observeOn(AndroidSchedulers.mainThread())
                    .doOnNext { logout(database) }
                    .doOnError { t -> t.printStackTrace() }

}

/**
 * Return true if autologin is needed
 */
class AutoLogin @Inject constructor(private val usersStorage: UsersStorage.LocalStorage) : BaseAction<Boolean, Nothing?>() {
    override fun prepareObservable(param: Nothing?): Observable<Boolean> = usersStorage.read().map { !it.password.isNullOrEmpty() }
}

class SignUp @Inject constructor(private val cloudStorage: UsersStorage.CloudStorage,
                                 private var usersStorage: UsersStorage.LocalStorage,
                                 private var tokenLocalStorage: SessionStorage.TokenLocalStorage) : RequestAction<LoginResponseModel, User>() {
    override fun prepareObservable(param: User): Observable<LoginResponseModel> {
        return cloudStorage.signUp(param)
                .doOnNext { ChatHelper.login(it.user.toQBUser()).subscribe(BaseDisposableObserver()) }
                .observeOn(AndroidSchedulers.mainThread())
                .doOnNext {
                    it.user.password = param.password; usersStorage.write(it.user)
                    tokenLocalStorage.write(Token(it.token))
                }
    }
}