在 Rx.NET 中，如何丰富（异步/可观察）查找中的值

Question

假设我有一个投标流 - 我想用投标人的名字来丰富它：

[
  { bidder: 'user/7', bet: 20  },
  { bidder: 'user/8', bet: 21 }, 
  { bidder: 'user/7', bet: 25  }, 
  /*...., 2 seconds later */
  { bidder: 'user/8', bet: 25  },
  { bidder: 'user/9', bet: 30  },
  ...

投标人名称来自网络服务：

GET '/users?id=7&id=8' =>
[{ user: 'user/7', name: 'Hugo Boss'}, { user: 'user/8', name: "Karl Lagerfeld"}

我将其包装到反应式读取缓存中：

IObservable<User> Users(IObservable<string> userIds);

撰写

现在我想将其组合成以下输出：

[
  { bidder: 'user/7', bet: 20, name: 'Hugo Boss'  },
  { bidder: 'user/8', bet: 21, name: 'Karl Lagerfeld' }, 
  { bidder: 'user/7', bet: 25, name: 'Hugo Boss'  }, 
  /*...., 2 seconds later */
  { bidder: 'user/8', bet: 25, name: 'Karl Lagerfeld'  },
  { bidder: 'user/9', bet: 30, name: 'Somebody else'  },
  ...

步骤 1

我想我需要将出价流投影到用户 ID 流。简单Select。然后在 read-through-cache 中，我用Buffer(TimeSpan, int) 将它分成块。

现在我有一个出价流和一个用户。

第二步

但是现在，如何将两者结合起来呢？

提示：对于 BID，保持订单是有意义的 - 但在我的真实代码中，我并不关心订单。所以我想要一个不依赖于用户缓存以正确顺序返回用户的解决方案。然后Zip 就可以完成这项工作。

我宁愿在获得用户信息后立即将所有出价发布到我的结果流中。

解决方案？

我很确定我需要以某种方式保持一些临时状态（窗口/缓冲区/...）。但我不知道在哪里以及如何。 也许这应该作为自定义运算符来实现；或者可能已经有一个了？

有什么想法吗？

编辑： 似乎不可能在流之上实际编写它。相反，我需要为 userid->user 函数获得一个承诺（Task 或 IObservable），并将其留给承诺批量加载和/或缓存用户，如果合适的话。

Answer 1

类似下面的东西应该可以工作。将缓存表示为 ISubject 并且您有一个异步缓存。在缓存内部，您可以在一个时间窗口内缓冲查询并将它们批量请求到服务器。

一些虚拟类

public struct User
{
    public string id;
}

public struct Bid
{
    public string userId;
    public int bid;
}

缓存对象本身

/// <summary>
/// Represent the cache by a subject that get's notified of
/// user requests and produces users. Internally this can
/// be buffered and chunked to the server. 
/// </summary>
public static ISubject<string, User> UsersCacheSubject;

分派和异步查询到数据库的方法

public static Task<User> UsersCache(string id)
{
    var r = UsersCacheSubject
        .Where(user=>user.id==id)
        .Replay(1)
        .Take(1);


    UsersCacheSubject.OnNext(id);

    return r.FirstAsync().ToTask();
}

试试看

 public void TryItOut()
{
    IObservable<Bid> bidderObservable = Observable.Repeat(new Bid());
     var foo = from bidder in bidderObservable
              from user in UsersCache(bidder.userId)
              where bidder.userId == user.id
              select new {bidder, user};
 }

Answer 2

这不就是像下面这样简单吗？

var query =
    from b in bids
    from u in Users(Observable.Return(b.Bidder))
    select new
    {
        b.Bidder,
        b.Bet,
        u.Name
    };