title: Delay and de-duplication using Reactive Extensions (Rx)
body: <p>I want to use Reactive Extensions to transform some messages and relay them after a small delay.</p>
  
  <p>The messages look something like this:</p>
  
  <pre><code>class InMsg
  {
     int GroupId { get; set; }
     int Delay { get; set; }
     string Content { get; set; }
  }
  </code></pre>
  
  <p>The output looks something like this:</p>
  
  <pre><code>class OutMsg
  { 
     int GroupId { get; set; }
     string Content { get; set; }
     OutMsg(InMsg in)
     {
         GroupId = in.GroupId;
         Content = Transform(in.Content);  // function omitted
     }
  }
  </code></pre>
  
  <p>There are a couple of requirements:</p>
  
  <ul>
  <li>The length of the delay is dependent on the content of the message.  </li>
  <li>Each message has a GroupId</li>
  <li>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.</li>
  </ul>
  
  <p>Given an Observable&lt;InMsg&gt; and a Send function:</p>
  
  <pre><code>IObservable&lt;InMsg&gt; inMsgs = ...;
  
  void Send(OutMsg o)
  {
       ... // publishes transformed messages
  }
  </code></pre>
  
  <p>I understand that I can use Select to perform the transformation.</p>
  
  <pre><code>void SetUp()
  {
       inMsgs.Select(i =&gt; new OutMsg(i)).Subscribe(Send);
  }
  </code></pre>
  
  <ul>
  <li>How can I apply a message specify delay? (Note this might/should result in out of order delivery of messages.)</li>
  <li>How can I de-dupe messages with the same GroupId?</li>
  <li>Is Rx capable of solving this problem?</li>
  <li>Is there another way of solving this?</li>
  </ul>