Циклические зависимости между потоками в реактивном программировании

Question

Погружаясь в реактивное программирование, я часто сталкиваюсь с ситуациями, когда два потока зависят друг от друга. Что такое идиоматический способ решения этих дел?

Минимальный пример: есть кнопки A и B, оба отображают значение. Нажатие на A должно увеличивать значение A на B. Нажатие на B должно установить значение B в A.

Первое решение, которое я мог бы придумать (например, в F #, но ответы на любом языке приветствуются):

let solution1 buttonA buttonB = 
    let mutable lastA = 0 
    let mutable lastB = 1 
    let a = new Subject<_>() 
    let b = new Subject<_>() 
    (OnClick buttonA).Subscribe(fun _ -> lastA <- lastA + lastB; a.OnNext lastA) 
    (OnClick buttonB).Subscribe(fun _ -> lastB <- lastA; b.OnNext lastB) 
    a.Subscribe(SetText buttonA) 
    b.Subscribe(SetText buttonA) 
    a.OnNext 0 
    b.OnNext 1 

Это решение использует изменчивое состояние и предметы, оно не очень читаемо и не выглядит идиоматическим.

Второе решение, которое я попытался включает в себя создание метода, который связывает два зависимых потоков вместе:

let dependency (aGivenB: IObservable<_> -> IObservable<_>) (bGivenA: IObservable<_> -> IObservable<_>) = 
    let bProxy = new ReplaySubject<_>() 
    let a = aGivenB bProxy 
    let b = bGivenA a 
    b.Subscribe(bProxy.OnNext) 
    a, b 

let solution2 buttonA buttonB = 
    let aGivenB b = 
     Observable.WithLatestFrom(OnClick buttonA, b, fun click bValue -> bValue) 
        .Scan(fun acc x -> acc + x) 
        .StartWith(0) 
    let bGivenA a = 
     Observable.Sample(a, OnClick buttonB) 
        .StartWith(1) 
    let a, b = dependency aGivenB bGivenA 
    a.Subscribe(SetText buttonA) 
    b.Subscribe(SetText buttonB) 

Это кажется немного лучше, но так как не существует способа, как dependency в реакционно-библиотеке, я считаю, что существуют более идиоматическое решение. Также легко ввести бесконечную рекурсию, используя второй подход.

Каков рекомендуемый способ решения проблем, связанных с циклической зависимостью между потоками, например, в приведенном выше примере, в реактивном программировании?

Answer 1

EDIT:

Вот F # Решение:

type DU = 
    | A 
    | B 

type State = { AValue : int; BValue : int } 

let solution2 (aObservable:IObservable<_>, bObservable:IObservable<_>) = 

    let union = aObservable.Select(fun _ -> A).Merge(bObservable.Select(fun _ -> B)) 

    let result = union.Scan({AValue = 0; BValue = 1}, fun state du -> match du with 
     | A -> { state with AValue = state.AValue + state.BValue } 
     | B -> { state with BValue = state.AValue } 
    ) 

    result 

---

F # на самом деле великий язык для этого, благодаря встроенному в дискриминированных союзами и записей. Вот ответ, написанный на C#, с пользовательским Дискриминационным Союзом; мой F # довольно ржавый.

Трюк состоит в том, чтобы превратить ваши наблюдаемые в одно наблюдаемое, используя дискриминированный союз. Таким образом, в основном, объединяющим а и Ь в одну наблюдаемую из дискриминационный союза:

a : *---*---*---** 
b : -*-*--*---*--- 
du: ab-ba-b-a-b-aa 

Как только это будет сделано, так что вы можете реагировать, если элемент является «А» толчок или толчок «B».

Просто для подтверждения, я предполагаю, что нет никакого способа явно указать значение значение, встроенное в ButtonA/ButtonB. Если есть, то эти изменения должны быть смоделированы как наблюдаемые, а также работать в дискриминационном союзе.

var a = new Subject<Unit>(); 
var b = new Subject<Unit>(); 
var observable = a.DiscriminatedUnion(b) 
    .Scan(new State(0, 1), (state, du) => du.Unify(
     /* A clicked case */_ => new State(state.A + state.B, state.B), 
     /* B clicked case */_ => new State(state.A, state.A) 
    ) 
); 

observable.Subscribe(state => Console.WriteLine($"a = {state.A}, b = {state.B}")); 
a.OnNext(Unit.Default); 
a.OnNext(Unit.Default); 
a.OnNext(Unit.Default); 
a.OnNext(Unit.Default); 
b.OnNext(Unit.Default); 
a.OnNext(Unit.Default); 
a.OnNext(Unit.Default); 
a.OnNext(Unit.Default); 
a.OnNext(Unit.Default); 
b.OnNext(Unit.Default); 

Вот классы, на которых это опирается на C#. Большая часть этого легко переводится во встроенные типы F #.

public class State /*easily replaced with an F# record */ 
{ 
    public State(int a, int b) 
    { 
     A = a; 
     B = b; 
    } 

    public int A { get; } 
    public int B { get; } 
} 

/* easily replaced with built-in discriminated unions and pattern matching */ 
public static class DiscriminatedUnionExtensions 
{ 
    public static IObservable<DiscriminatedUnionClass<T1, T2>> DiscriminatedUnion<T1, T2>(this IObservable<T1> a, IObservable<T2> b) 
    { 
     return Observable.Merge(
      a.Select(t1 => DiscriminatedUnionClass<T1, T2>.Create(t1)), 
      b.Select(t2 => DiscriminatedUnionClass<T1, T2>.Create(t2)) 
     ); 
    } 

    public static IObservable<TResult> Unify<T1, T2, TResult>(this IObservable<DiscriminatedUnionClass<T1, T2>> source, 
     Func<T1, TResult> f1, Func<T2, TResult> f2) 
    { 
     return source.Select(union => Unify(union, f1, f2)); 
    } 

    public static TResult Unify<T1, T2, TResult>(this DiscriminatedUnionClass<T1, T2> union, Func<T1, TResult> f1, Func<T2, TResult> f2) 
    { 
     return union.Item == 1 
      ? f1(union.Item1) 
      : f2(union.Item2) 
     ; 
    } 
} 

public class DiscriminatedUnionClass<T1, T2> 
{ 
    private readonly T1 _t1; 
    private readonly T2 _t2; 
    private readonly int _item; 
    private DiscriminatedUnionClass(T1 t1, T2 t2, int item) 
    { 
     _t1 = t1; 
     _t2 = t2; 
     _item = item; 
    } 

    public int Item 
    { 
     get { return _item; } 
    } 

    public T1 Item1 
    { 
     get { return _t1; } 
    } 

    public T2 Item2 
    { 
     get { return _t2; } 
    } 

    public static DiscriminatedUnionClass<T1, T2> Create(T1 t1) 
    { 
     return new DiscriminatedUnionClass<T1, T2>(t1, default(T2), 1); 
    } 

    public static DiscriminatedUnionClass<T1, T2> Create(T2 t2) 
    { 
     return new DiscriminatedUnionClass<T1, T2>(default(T1), t2, 2); 
    } 
} 

Answer 2

Вот очень простое решение с использованием Gjallarhorn:

#r @"..\packages\Gjallarhorn\lib\portable-net45+netcore45+wpa81+wp8+MonoAndroid1+MonoTouch1\Gjallarhorn.dll" 

open Gjallarhorn 

(* 
    Clicking on A must increment the value of A by B. Clicking on B must set the value of B to A. 
*) 
let a = Mutable.create 3 
let b = Mutable.create 4 

let clickA() = a.Value <- a.Value + b.Value 
let clickB() = b.Value <- a.Value 

let d1 = Signal.Subscription.create (fun x -> printfn "%A" <| "Clicked A: " + x.ToString()) a 
let d2 = Signal.Subscription.create (fun x -> printfn "%A" <| "Clicked B: " + x.ToString()) b 

clickA() 
clickB() 

Это на самом деле очень похоже на ваш первоначальный так же использовать изменяемое состояние, но делает связывание с UI довольно легко, для более идиоматических использования посмотреть blog post ,

Answer 3

Предполагая, что выход в конечном итоге будет отправлен обратно в источник, вы можете сделать это с помощью основных операторов. Все, что вам нужно сделать, это позвонить withLatestFrom дважды за каждую кнопку/сигнал, наблюдаемый. Мое решение находится в java, но должно быть достаточно легко следовать!

private static Pair<Observable<Integer>, Observable<Integer>> test(
    final Observable<Integer> aValues, 
    final Observable<Integer> bValues, 
    final Observable<Void> aButton, 
    final Observable<Void> bButton, 
    final Func2<Integer, Integer, Integer> aFunction, 
    final Func2<Integer, Integer, Integer> bFunction 
) { 
    return new Pair<>(
     aButton.withLatestFrom(aValues, (button, a) -> a).withLatestFrom(bValues, aFunction), 
     bButton.withLatestFrom(aValues, (button, a) -> a).withLatestFrom(bValues, bFunction) 
    ); 
} 

Heres проверочный код, я использовал:

final TestScheduler scheduler = new TestScheduler(); 

final TestSubject<Integer> aSubject = TestSubject.create(scheduler); 
final TestSubject<Integer> bSubject = TestSubject.create(scheduler); 
aSubject.onNext(1); 
bSubject.onNext(1); 

final TestSubject<Void> aButton = TestSubject.create(scheduler); 
final TestSubject<Void> bButton = TestSubject.create(scheduler); 

final Pair<Observable<Integer>, Observable<Integer>> pair = test(
    aSubject, bSubject, aButton, bButton, (a, b) -> a + b, (a, b) -> a 
); 

pair.component1().subscribe(aSubject::onNext); 
pair.component2().subscribe(bSubject::onNext); 
pair.component1().map(a -> "A: " + a).subscribe(System.out::println); 
pair.component2().map(b -> "B: " + b).subscribe(System.out::println); 

aButton.onNext(null); scheduler.triggerActions(); 
bButton.onNext(null); scheduler.triggerActions(); 
aButton.onNext(null); scheduler.triggerActions(); 
aButton.onNext(null); scheduler.triggerActions(); 
bButton.onNext(null); scheduler.triggerActions(); 

Печатается:

A: 2 
B: 2 
A: 4 
A: 6 
B: 6