使用反应性扩展 (Rx) 进行延迟和重复数据删除

Question

我想使用反应式扩展来转换一些消息并在小延迟后中继它们。

消息看起来像这样：

class InMsg
{
   int GroupId { get; set; }
   int Delay { get; set; }
   string Content { get; set; }
}

输出看起来像这样：

class OutMsg
{ 
   int GroupId { get; set; }
   string Content { get; set; }
   OutMsg(InMsg in)
   {
       GroupId = in.GroupId;
       Content = Transform(in.Content);  // function omitted
   }
}

有几个要求：

• 延迟的长度取决于消息的内容。
• 每个消息都有一个 GroupId
• 如果新消息与等待传输的延迟消息具有相同的 GroupId，则应丢弃第一个消息，并在新的延迟周期后仅传输第二个消息。

给定一个 Observable和 Send 函数：

IObservable<InMsg> inMsgs = ...;

void Send(OutMsg o)
{
     ... // publishes transformed messages
}

我知道我可以使用 Select 来执行转换。

void SetUp()
{
     inMsgs.Select(i => new OutMsg(i)).Subscribe(Send);
}

• 如何应用指定延迟的消息？ （请注意，这可能/应该会导致消息传送无序。）
• 如何对具有相同 GroupId 的消息进行重复数据删除？
• Rx能解决这个问题吗？
• 还有其他方法可以解决这个问题吗？

Answer 1

您可以使用GroupBy来创建IGroupedObservable，使用Delay来延迟输出，使用Switch来确保更新的值替换组中以前的值：

IObservable<InMsg> inMessages;

inMessages
    .GroupBy(msg => msg.GroupId)
    .Select(group =>
        {
            return group.Select(groupMsg => 
                {
                    TimeSpan delay = TimeSpan.FromMilliseconds(groupMsg.Delay);
                    OutMsg outMsg = new OutMsg(); // map InMsg -> OutMsg here

                    return Observable.Return(outMsg).Delay(delay);
                })
                .Switch();
        })
        .Subscribe(outMsg => Console.Write("OutMsg received"));

关于实现的注意事项：如果分组值在发送消息之后（即延迟之后）到达，它将启动新的延迟

Answer 2

@Richard Szalay 的答案几乎对我有用（在.NET Framework 4.6上使用.NET Rx 3.1.1），但我必须在末尾添加.Merge()用于组合 IObservable> 结果的表达式，如下所示：

对于我（在 .NET Framework 4.6 上使用 .NET Rx 3.1.1），修复方法是添加 。 Merge() 到最后，如下所示：

var deduplicated = inputs
    .GroupBy(input => input)
    .Select(group =>
        group
        .Select(input => Observable.Return(input).Delay(TimeSpan.FromSeconds(5)))
        .Switch())
    .Merge(); // <-- This is added to combine the partitioned results