ExternalVendorCode/corda
1
0
Fork 0
mirror of https://github.com/corda/corda.git synced 2025-04-08 03:44:49 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Update kdocs of FlowSafeSubject/ PreventSubscriptionsSubject

Browse Source
This commit is contained in:
Kyriakos Tharrouniatis 2020-02-16 20:30:41 +00:00
parent a194c79e5f
commit d341a982a3
1 changed files with 6 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
core/src/main/kotlin/net/corda/core/utilities/FlowSafeSubject.kt
View File

@ -4,17 +4,16 @@ import net.corda.core.internal.FlowSafeSubscriber
import net.corda.core.internal.VisibleForTesting
import rx.Observable.OnSubscribe
import rx.Observer
import rx.Subscriber
import rx.observers.SafeSubscriber
import rx.subjects.Subject
/**
* The [FlowSafeSubject] is used to unwrap a [SafeSubscriber] to prevent the observer from unsubscribing from the base observable when any
* error occurs. Calls to [SafeSubscriber._onError] call [Subscriber.unsubscribe] multiple times, which stops the observer receiving updates
* from the observable.
* [FlowSafeSubject] is used to unwrap an [Observer] from a [SafeSubscriber], re-wrap it with a [FlowSafeSubscriber]
* and then subscribe it to its underlying [Subject]. It is only used to provide therefore, its underlying [Subject] with
* non unsubscribing [rx.Observer]s.
*
* Preventing this is useful to observers that are subscribed to important observables to prevent them from ever unsubscribing due to an
* error. Unsubscribing could lead to a malfunctioning CorDapp, for the rest of the current run time, due to a single isolated error.
* Upon [rx.Observable.subscribe] it will wrap everything that is a non [SafeSubscriber] with a [FlowSafeSubscriber] the same way
* [rx.subjects.PublishSubject] wraps everything that is a non [SafeSubscriber] with a [SafeSubscriber].
*/
@VisibleForTesting
class FlowSafeSubject<T, R>(private val actual: Subject<T, R>) : Observer<T> by actual,
@ -32,7 +31,7 @@ class FlowSafeSubject<T, R>(private val actual: Subject<T, R>) : Observer<T> by
}
/**
* The [PreventSubscriptionsSubject] is used to prevent any subscriptions to a [Subject].
* The [PreventSubscriptionsSubject] is used to prevent any subscriptions to its underlying [Subject].
*/
class PreventSubscriptionsSubject<T, R>(private val actual: Subject<T, R>, errorAction: () -> Unit) : Observer<T> by actual,
Subject<T, R>(OnSubscribe<R> { _ ->