Edit Question - KnowledgeHub

Title Body

I want to use Reactive Extensions to transform some messages and relay them after a small delay. The messages look something like this: class InMsg { int GroupId { get; set; } int Delay { get; set; } string Content { get; set; } } The output looks something like this: class OutMsg { int GroupId { get; set; } string Content { get; set; } OutMsg(InMsg in) { GroupId = in.GroupId; Content = Transform(in.Content); // function omitted } } There are a couple of requirements: The length of the delay is dependent on the content of the message. Each message has a GroupId If a newer message comes in with the same GroupId as a delayed message awaiting transmission then the first message should be dropped and only the second one transmitted after a new delay period. Given an Observable<InMsg> and a Send function: IObservable<InMsg> inMsgs = ...; void Send(OutMsg o) { ... // publishes transformed messages } I understand that I can use Select to perform the transformation. void SetUp() { inMsgs.Select(i => new OutMsg(i)).Subscribe(Send); } How can I apply a message specify delay? (Note this might/should result in out of order delivery of messages.) How can I de-dupe messages with the same GroupId? Is Rx capable of solving this problem? Is there another way of solving this?

Tags (comma-separated) Cancel