Delay and de-duplication using Reactive Extensions (Rx)

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?