c# - 使用 CombineLatest 关联多个组的三个事件流

Question

我有 3 个可观察的事件流 - 按日期顺序 -(主要事件、有关销售的事件和有关客户的事件) - 每个流都包含一种与各种车辆相关的事件，每个事件都有一个 vehicleID 和各种其他事件特性。事件可以出现在一辆车上，然后出现在另一辆车上等等。所以基本上我试图根据 VehicleID 将三个独立的事件流关联在一起——这听起来应该很简单。我对任何形式的复杂可观察编程都不熟悉，所以这证明相当困难。

每当我在任何流中看到车辆的新事件时，我都想调用一个函数(我想基本上是 combineLatest)。如果我过滤每个流只包含一辆车的事件，我可以做到这一点，所以 Where，但我不知道如何 GroupBy 然后获取最新的每组。我想我正在寻求合并流，但在每组车辆上结合最新。

下面将打印我希望为 VehcileID=1 创建的所有对象。我希望对所有车辆执行以下操作。如果我用每个 VehcileID 循环遍历它，这将给我我想要的输出 - 但这看起来不像是时髦的观察 - 一切都是流 - 我应该瞄准的禅宗状态。

Observable.CombineLatest(mainEvents.Where(a=>a.VehcileID==1),saleEventsGroup.Where(a=>a.VehcileID==1),customerEventsGroup.Where(a=>a.VehcileID==1),(main, sale, customer)=>{ 
        //Basically flattening various properties from latest state of the 3 streams for current vehicle with some mapping
        return ComplexObject(){};})       
        .Subscribe(Console.WriteLine);

如何为每辆车的每个流合并最新事件。

如有任何建议，我们将不胜感激

Answer 1

这个怎么样？我在这里只做两个流，但这个想法可以很容易地扩展到三个流

   [TestMethod]
   public void GroupByWithMultipleStreams()
    {
        Subject<Notification> producer = new Subject<Notification>();
        Subject<RelatedToNotification> otherThingProducer = new Subject<RelatedToNotification>();            

        Observable.Merge(
            producer.Select(n => new { Id = n.Id, notification = n, relatedNotification = (RelatedToNotification)null }),
            otherThingProducer.Select(rn => new { Id = rn.NotificationId, notification = (Notification)null, relatedNotification = rn }))
            .GroupBy(x => x.Id)
            .SelectMany(obs =>
            {
                return obs.Scan(new ComplexObject() { Id = obs.Key }, (acc, input) =>
                {
                    acc.Notification = input.notification ?? acc.Notification;
                    acc.Related = input.relatedNotification ?? acc.Related;
                    return acc;
                });
            })
            .Where(result => result.Notification != null && result.Related != null) // if you only want it to fire when everything has a value
            .Subscribe(result =>
            {
                //do something with the results here
            }
            );

        producer.OnNext(new Notification() { Id = 1, Version = 1 });
        producer.OnNext(new Notification() { Id = 1, Version = 2 });
        producer.OnNext(new Notification() { Id = 2, Version = 17 });
        producer.OnNext(new Notification() { Id = 1, Version = 3 });
        producer.OnNext(new Notification() { Id = 9, Version = 0 });
        producer.OnNext(new Notification() { Id = 9, Version = 1 });
        otherThingProducer.OnNext(new RelatedToNotification() { NotificationId = 2,  SomeData = "2data" });
        otherThingProducer.OnNext(new RelatedToNotification() { NotificationId = 2, SomeData = "2data1" });
        otherThingProducer.OnNext(new RelatedToNotification() { NotificationId = 9, SomeData = "9Data" });
        producer.OnNext(new Notification() { Id = 2, Version = 1 });

    }

    class ComplexObject
    {
        public int Id { get; set; }
        public Notification Notification { get; set; }
        public RelatedToNotification Related { get; set; }
    }

    class Notification
    {
        public int Id { get; set; }
        public int Version { get; set; }

        public string Name { get; set; }
    }

    public class RelatedToNotification
    {
        public int NotificationId { get; set; }
        public string SomeData { get; set; }
    }