AAverin

class ResolvedSubscriber<T> constructor(
        val resolution: Resolution,
        val onNextFunc: (T) -> Unit,
        val onCompletedFunc: () -> Unit = {},
        val onErrorFunc: (Throwable?) -> Unit = {}) : Subscriber<T>() {

    override fun onNext(p0: T) {
        if (!isUnsubscribed) {
            onNextFunc(p0)
        }

AAverin

open class UIResolution @Inject constructor(val uiResolver: UIResolver) : ResolutionByCase() {

    override fun onConnectivityAvailable() {
        uiResolver.hidePersistentSnackBar()
    }

    override fun onConnectivityUnavailable() {
        uiResolver.showPersistentSnackBar(R.string.error_no_network_connection)
    }

AAverin

interface RxHttpResolution {
    fun onHttpException(httpException: HttpException)
    fun onGenericRxException(t: Throwable)
}

interface NetworkConnectivityResolution {
    fun onConnectivityAvailable()
    fun onConnectivityUnavailable()
}

AAverin

@RunWith(PowerMockRunner::class)
@PrepareForTest(RxBus::class, CompositeSubscription::class, Observable::class)
class BusSubscriberTest {

    @Mock lateinit var subscription: Subscription
    @Mock lateinit var bus: RxBus
    @Mock lateinit var observable: Observable<Any>

    @Captor var subscriptionCaptor = ArgumentCaptor.forClass(Subscription::class.java)

AAverin

interface BusSubscriberContract {
    fun bus(bus: RxBus)
    fun <T : Any> subscribe(clazz: KClass<T>, callback: (T) -> Unit)
    fun unsubscribe()
}

open class BusSubscriber @Inject constructor() : BusSubscriberContract {

    val subscriptions = mutableListOf<Subscription>()

AAverin

@Singleton
class GlobalBus @Inject constructor() : RxBus()

AAverin

open class RxBus {
    private val bus = SerializedSubject<BusEvent, BusEvent>(PublishSubject.create());

    fun post(event: BusEvent) {
        bus.onNext(event)
    }

    fun <T> events(type: Class<T>): Observable<T> {
        return events().ofType(type)
    }

AAverin

import rx.Observable
import rx.Observer
import rx.Subscriber
import rx.Subscription
import rx.functions.Action0
import rx.functions.Action1

fun <T> Observable<T>.uiSubscribe(schedulers: Schedulers, subscriber: Subscriber<in T>): Subscription {
    return subscribeOn(schedulers.io)
            .observeOn(schedulers.mainThread)

AAverin

interface ConsistencyManagerContract {
    fun notifyUpdateModel(model: Any)
    fun notifyUpdateClass(kclass: KClass<*>)
    fun subscribe(modelClass: KClass<*>, callback: WeakReference<(Any?) -> Unit>)
}

class ConsistencyManager : ConsistencyManagerContract {
    private val subscriptionsMap: MutableMap<KClass<*>, MutableList<WeakReference<(Any?) -> Unit>>> = mutableMapOf()

    override fun notifyUpdateModel(model: Any) {

AAverin

interface ConsistencyManagerContract {
    fun notifyUpdateModel(model: Any)
    fun notifyUpdateClass(kclass: KClass<*>)
    fun subscribe(modelClass: KClass<*>, callback: WeakReference<(Any?) -> Unit>)
}

class ConsistencyManager : ConsistencyManagerContract {
    private val subscriptionsMap: MutableMap<KClass<*>, MutableList<WeakReference<(Any?) -> Unit>>> = mutableMapOf()

    override fun notifyUpdateModel(model: Any) {