2つの観察可能なシーケンスを一致するキーでペアリングすることはできますか？

Question

この質問は2つの違いで、同様の名称質問hereに基づいています。

• を私は複数のキーを上に一致しています。問題ない。
• キーが繰り返されることがあります。問題。

私のテストコードは以下の通りです。

• CoordBundleが、すぐに観察された少なくとも1つのCoordMetricsと1 CoordDataとして公開されている：私は、次の動作を必要としています。
• 特定のX/Yキーがいずれかのobservableで繰り返されると、新しいCoordBundleが公開されます。

これを行うにはどうすればよいですか？

public class CoordMetrics 
{ 
    internal CoordMetrics(int x, int y, IEnumerable<IMetric> metrics) 
    { 
     X = x; 
     Y = y; 
     Metrics = metrics; 
    } 
    internal int X { get; private set; } 
    internal int Y { get; private set; } 
    internal IEnumerable<IMetric> Metrics { get; private set; } 
} 

public class CoordData 
{ 
    internal CoordData(int x, int y, IEnumerable<IDatum> data) 
    { 
     X = x; 
     Y = y; 
     Data = data; 
    } 

    internal int X { get; private set; } 
    internal int Y { get; private set; } 
    internal IEnumerable<IDatum> Data { get; private set; } 
} 

public class CoordBundle 
{ 
    internal CoordBundle(int x, int y, IEnumerable<IMetric> metrics, IEnumerable<IDatum> data) 
    { 
     X = x; 
     Y = y; 
     Metrics = metrics; 
     Data = data; 
    } 

    internal int X { get; private set; } 
    internal int Y { get; private set; } 
    internal IEnumerable<IMetric> Metrics { get; private set; } 
    internal IEnumerable<IDatum> Data { get; private set; } 
} 

[TestClass] 
public class PairingTest 
{ 
    [TestMethod, TestCategory("Temp")] 
    public void PairedObservableTest() 
    { 
     Trace.Listeners.Add(new TextWriterTraceListener(Console.Out)); 
     var aSource = new Subject<CoordMetrics>(); 
     var bSource = new Subject<CoordData>(); 

     var paired = Observable.Merge(aSource.Select(a => new Pair(a, null)), bSource.Select(b => new Pair(null, b))) 
           .GroupBy(p => p.Item1 != null ? new { p.Item1.X, p.Item1.Y } : new { p.Item2.X, p.Item2.Y }) 
           .SelectMany(g => g.Buffer(2).Take(1)) 
           .Select(g => new Pair(
            g.ElementAt(0).Item1 ?? g.ElementAt(1).Item1, 
            g.ElementAt(0).Item2 ?? g.ElementAt(1).Item2)) 
           .Select(t => new CoordBundle(t.Item1.X, t.Item1.Y, t.Item1.Metrics, t.Item2.Data)); 

     paired.Subscribe(g => Trace.WriteLine(String.Format("{0},{1}", g.X, g.Y))); 

     bSource.OnNext(new CoordData(2, 1, Enumerable.Empty<IDatum>())); 
     aSource.OnNext(new CoordMetrics(2, 2, Enumerable.Empty<IMetric>())); 
     aSource.OnNext(new CoordMetrics(1, 1, Enumerable.Empty<IMetric>())); 
     bSource.OnNext(new CoordData(1, 2, Enumerable.Empty<IDatum>())); 
     bSource.OnNext(new CoordData(2, 2, Enumerable.Empty<IDatum>())); 
     bSource.OnNext(new CoordData(1, 1, Enumerable.Empty<IDatum>())); 
     aSource.OnNext(new CoordMetrics(1, 2, Enumerable.Empty<IMetric>())); 
     aSource.OnNext(new CoordMetrics(2, 1, Enumerable.Empty<IMetric>())); 
     aSource.OnNext(new CoordMetrics(2, 2, Enumerable.Empty<IMetric>())); 
     bSource.OnNext(new CoordData(2,2,Enumerable.Empty<IDatum>())); 
    } 
} 

所望の出力 - 出力上記のコードのみ最初の4行：

2,2 
1,1 
1,2 
2,1 
2,2 
2,2 

Answer 1

私は私はあなたが欲しいものを持っていると思います。 これは簡単な問題ではありません。 一方の配列が他方の配列であることはかなり一般的ですが、ここではどちらかの配列が他方の配列になることがあります。

実用的な解決策を得るために私がやった最初のことは、検証可能な単体テストに分解することでした。 TestSchedulerとこれに関連するタイプを使用して（件名などの代わりに）これを行うことをお勧めします。

私はあなたの要件がだと思うものから大理石の図を作成しました。 次に、2つのテスト入力シーケンスと期待される出力シーケンスにマップすることができました。

最後の部分は実際にクエリを作成することでした。

*で終わったアプローチは、マスターシーケンスと子シーケンスから一致するようにしようとする2つのシーケンスを作成することでした - >SelectMany + Whereです。 しかし、両方の入力がマスターシーケンスの役割を果たすことができるので、私はこれを2回行う必要がありました。 2度購読するので、シーケンスを共有する必要がありました - >Publish()。 また、各シーケンスが複数の値を生成する可能性があるため、複製が到着したときに以前の一致からの一致を取り消す必要がありました - >TakeUntil。最後に、2つの結果セットをまとめて「Merge」にまとめました。

*私はGroupJoin & CombineLatestと考えていましたが、彼らは私にとってはうまくいかなかったようです。

[TestClass] 
public class PairingTest 
{ 
    [TestMethod, TestCategory("Temp")] 
    public void PairedObservableTest() 
    { 
     var scheduer = new TestScheduler(); 

     /* 
     Legend 
      a = aSource (CoordMetrics) 
      b = bSource (CoordData) 
      r = expected result 


     a ----2--1-----------1--2--2----- 
       2 1   2 1 2 

     b -2--------1--2--1-----------2-- 
      1  2 2 1   2 

     r -------------2--1--1--2--2--2-- 
         2 1 2 1 2 2 
     */ 
     var aSource = scheduer.CreateColdObservable<CoordMetrics>(
      ReactiveTest.OnNext(5, new CoordMetrics(2, 2)), 
      ReactiveTest.OnNext(8, new CoordMetrics(1, 1)), 
      ReactiveTest.OnNext(20, new CoordMetrics(1, 2)), 
      ReactiveTest.OnNext(23, new CoordMetrics(2, 1)), 
      ReactiveTest.OnNext(26, new CoordMetrics(2, 2)) 
     ); 
     var bSource = scheduer.CreateColdObservable<CoordData>(
      ReactiveTest.OnNext(2, new CoordData(2, 1)), 
      ReactiveTest.OnNext(11, new CoordData(1, 2)), 
      ReactiveTest.OnNext(14, new CoordData(2, 2)), 
      ReactiveTest.OnNext(17, new CoordData(1, 1)), 
      ReactiveTest.OnNext(29, new CoordData(2, 2)) 
     ); 

     var testObserver = scheduer.CreateObserver<string>(); 
     Implementation(aSource, bSource) 
      .Subscribe(testObserver); 



     scheduer.Start(); 

     ReactiveAssert.AreElementsEqual(
      new[] { 
        ReactiveTest.OnNext(14, "2,2"), 
        ReactiveTest.OnNext(17, "1,1"), 
        ReactiveTest.OnNext(20, "1,2"), 
        ReactiveTest.OnNext(23, "2,1"), 
        ReactiveTest.OnNext(26, "2,2"), 
        ReactiveTest.OnNext(29, "2,2") 
       }, 
      testObserver.Messages 
     ); 
    } 

    private static IObservable<string> Implementation(IObservable<CoordMetrics> aSource, IObservable<CoordData> bSource) 
    { 
     return Observable.Create<string>(observer => 
     { 
      var aShared = aSource.Publish(); 
      var bShared = bSource.Publish(); 

      var fromA = aShared.SelectMany(a => bShared 
        //Find matching values from B's 
        .Where(b => a.X == b.X && a.Y == b.Y) 
        //Only run until another matching A is produced 
        .TakeUntil(aShared.Where(a2 => a2.X == a.X && a2.Y == a.Y)) 
        //Project/Map to required type. 
        .Select(b => new CoordBundle(a.X, a.Y /*, a.Metrics, b.Data*/)) 
       ); 

      var fromB = bShared.SelectMany(b => aShared 
        //Find matching values from A's 
        .Where(a => a.X == b.X && a.Y == b.Y) 
        //Only run until another matching B is produced 
        .TakeUntil(bShared.Where(b2 => b2.X == b.X && b2.Y == b.Y)) 
        //Project/Map to required type. 
        .Select(a => new CoordBundle(a.X, a.Y /*, a.Metrics, b.Data*/)) 
       ); 

      var paired = Observable.Merge(fromA, fromB); 

      paired 
       .Select(g => String.Format("{0},{1}", g.X, g.Y)) 
       .Subscribe(observer); 

      return new CompositeDisposable(aShared.Connect(), bShared.Connect()); 
     }); 
    } 
} 

// Define other methods and classes here 
public class CoordMetrics 
{ 
    internal CoordMetrics(int x, int y) 
    { 
     X = x; 
     Y = y; 
    } 
    internal int X { get; private set; } 
    internal int Y { get; private set; } 
} 

public class CoordData 
{ 
    internal CoordData(int x, int y) 
    { 
     X = x; 
     Y = y; 
    } 

    internal int X { get; private set; } 
    internal int Y { get; private set; } 
} 

public class CoordBundle 
{ 
    internal CoordBundle(int x, int y) 
    { 
     X = x; 
     Y = y; 
    } 

    internal int X { get; private set; } 
    internal int Y { get; private set; } 
} 
public class Pair 
{ 
    public Pair(CoordMetrics x, CoordData y) 
    { 
     Item1 = x; 
     Item2 = y; 
    } 
    public CoordMetrics Item1 { get; set; } 
    public CoordData Item2 { get; set; } 
}