反应式编程中的流之间的循环依赖关系

Question

在反应式编程中涉猎很多时，我经常会遇到两个流互相依赖的情况。什么是解决这些案件的惯用方法？

一个简单的例子：有按钮A和B，都显示一个值。点击A必须使B的值增加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

编辑：

这里有一个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＃很生锈。

诀窍是使用区分的联合将您的两个observable转换为一个observable。所以基本上团结a和b成一个可观察鉴别联合：

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

一旦做到这一点，这样你就可以到，如果该项目是一个“A”推或“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() 

它实际上非常相似，你的初始所以没有使用可变的状态，但使绑定到用户界面非常简单，更多的习惯用法看到这个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) 
    ); 
} 

继承人的测试代码我使用：

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