title: Rx operator to distinct sequences
body: <p><strong>IMPORTANT</strong>: for a description of the results and some more details, please have a look also to my answer</p>
  
  <p>I need to group and filter a sequence of objects/events that usually are replicated, buffering them with a TimeSpan interval. I try to explain it better with sort of marble diagrams:</p>
  
  <pre><code>X-X-X-X-X-Y-Y-Y-Z-Z-Z-Z-X-X-Y-Z-Z
  </code></pre>
  
  <p>would produce</p>
  
  <pre><code>X---Y---Z---X---Y---Z
  </code></pre>
  
  <p>where X, Y and Z are different event types, and '---' means the interval.
  Additionally, I would also like to distinct by a key property that it is available on all types because they have a common base class:</p>
  
  <pre><code>X, Y, Z : A
  </code></pre>
  
  <p>and A contains a property Key. Using the notation X.a meaning X.Key = a, A final sample would be:</p>
  
  <pre><code>X.a-X.b-X.a-Y.b-Y.c-Z.a-Z.a-Z.c-Z.b-Z.c
  </code></pre>
  
  <p>would produce</p>
  
  <pre><code>X.a-X.b---Y.b-Y.c-Z.a-Z.c-Z.b
  </code></pre>
  
  <p>Can anybody help me putting together the required Linq operators (probably DistinctUntilChanged and Buffer) to achieve this behavior?
  Thanks</p>
  
  <p><strong>UPDATE 18.08.12</strong>:</p>
  
  <p>as requested, I try to give a better explanation.
  We have devices collecting and sending events to a web service. These devices have an old logic (and we can't change it due to backward compatibility) and they continuously send an event until they receive an acknowledge; after the acknowledge, they send the next event in their queue, and so on.
  Events contain the network address of the unit and some other properties distinguishing events in the queue for each device.
  An event looks like this:</p>
  
  <pre><code>class Event
  {
      public string NetworkAddress { get; }
  
      public string EventCode { get; }
  
      public string AdditionalAttribute { get; }
  }
  </code></pre>
  
  <p>The goal is that of processing every 5 seconds the distinguished events received from all devices, storing information in the database (that's why we don't want to do it in batches) and sending the ack to the device.
  Let's make an example with only two devices and some events:</p>
  
  <pre><code>Device 'a':
  Event 1 (a1): NetworkAddress = '1', EventCode = A, AdditionalAttribute = 'x'
  Event 2 (a2): NetworkAddress = '1', EventCode = A, AdditionalAttribute = 'y'
  Event 3 (a3): NetworkAddress = '1', EventCode = B, AdditionalAttribute = 'x'
  
  Device 'b':
  Event 1 (b1): NetworkAddress = '2', EventCode = A, AdditionalAttribute = 'y'
  Event 2 (b2): NetworkAddress = '2', EventCode = B, AdditionalAttribute = 'x'
  Event 3 (b3): NetworkAddress = '2', EventCode = B, AdditionalAttribute = 'y'
  Event 4 (b4): NetworkAddress = '2', EventCode = C, AdditionalAttribute = 'x'
  
  Pn are the operations done by our server, explained later
  </code></pre>
  
  <p>Possible marble diagram (input streams + output stream):</p>
  
  <pre><code>Device 'a'          : -[a1]-[a1]-[a1]----------------[a2]-[a2]-[a2]-[a3]-[a3]-[a3]-...
  Device 'b'          : ------[b1]-[b1]-[b2]-[b2]-[b2]------[b3]-[b3]-[b4]-[b4]-[b4]-...
  
  Time                : ------------[1s]-----------[2s]------------[3s]------------[4s]-
  DB/acks (rx output) : ------------[P1]-----------[P2]------------[P3]------------[P4]-
  
  P1: Server stores and acknowledges [a1] and [b1]
  P2: "      "      "   "            [b2]
  P3: "      "      "   "            [a2] and [b3]
  P4: "      "      "   "            [a3] and [b4]
  </code></pre>
  
  <p>At the end I think it is probably a simple combination of basic operators, but I'm new to Rx and I'm a bit confused since it seems that there are lots of operators (or combinations of operators) to get the same output stream.</p>
  
  <p><strong>Update 19.08.12</strong>:</p>
  
  <p>Please keep in mind that this code runs on a server and it should run for days without memory leaks...I'm not sure about the behavior of subjects. At the moment, for each event I call a push operation on a service, which calls the OnNext of a Subject on top of which I should build the query (if I'm not wrong about the usage of subjects).</p>
  
  <p><strong>Update 20.08.12</strong>:</p>
  
  <p>Current implementation, including validation test; this is what I tried and it seems the same suggested by @yamen</p>
  
  <pre><code>public interface IEventService
  {
      // Persists the events
      void Add(IEnumerable&lt;Event&gt; events);
  }
  
  public class Event
  {
      public string Description { get; set; }
  }
  
  /// &lt;summary&gt;
  /// Implements the logic to handle events.
  /// &lt;/summary&gt;
  public class EventManager : IDisposable
  {
      private static readonly TimeSpan EventHandlingPeriod = TimeSpan.FromSeconds(5);
  
      private readonly Subject&lt;EventMessage&gt; subject = new Subject&lt;EventMessage&gt;();
  
      private readonly IDisposable subscription;
  
      private readonly object locker = new object();
  
      private readonly IEventService eventService;
  
      /// &lt;summary&gt;
      /// Initializes a new instance of the &lt;see cref="EventManager"/&gt; class.
      /// &lt;/summary&gt;
      /// &lt;param name="scheduler"&gt;The scheduler.&lt;/param&gt;
      public EventManager(IEventService eventService, IScheduler scheduler)
      {
          this.eventService = eventService;
          this.subscription = this.CreateQuery(scheduler);
      }
  
      /// &lt;summary&gt;
      /// Pushes the event.
      /// &lt;/summary&gt;
      /// &lt;param name="eventMessage"&gt;The event message.&lt;/param&gt;
      public void PushEvent(EventMessage eventMessage)
      {
          Contract.Requires(eventMessage != null);
          this.subject.OnNext(eventMessage);
      }
  
      /// &lt;summary&gt;
      /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
      /// &lt;/summary&gt;
      /// &lt;filterpriority&gt;2&lt;/filterpriority&gt;
      public void Dispose()
      {
          this.Dispose(true);
      }
  
      private void Dispose(bool disposing)
      {
          if (disposing)
          {
              // Dispose unmanaged resources
          }
  
          this.subject.Dispose();
          this.subscription.Dispose();
      }
  
      private IDisposable CreateQuery(IScheduler scheduler)
      {
          var buffered = this.subject
              .DistinctUntilChanged(new EventComparer())
              .Buffer(EventHandlingPeriod, scheduler);
  
          var query = buffered
              .Subscribe(this.HandleEvents);
          return query;
      }
  
      private void HandleEvents(IList&lt;EventMessage&gt; eventMessages)
      {
          Contract.Requires(eventMessages != null);
          var events = eventMessages.Select(this.SelectEvent);
          this.eventService.Add(events);
      }
  
      private Event SelectEvent(EventMessage message)
      {
          return new Event { Description = "evaluated description" };
      }
  
      private class EventComparer : IEqualityComparer&lt;EventMessage&gt;
      {
          public bool Equals(EventMessage x, EventMessage y)
          {
              return x.NetworkAddress == y.NetworkAddress &amp;&amp; x.EventCode == y.EventCode &amp;&amp; x.Attribute == y.Attribute;
          }
  
          public int GetHashCode(EventMessage obj)
          {
              var s = string.Concat(obj.NetworkAddress + "_" + obj.EventCode + "_" + obj.Attribute);
              return s.GetHashCode();
          }
      }
  }
  
  public class EventMessage
  {
      public string NetworkAddress { get; set; }
  
      public byte EventCode { get; set; }
  
      public byte Attribute { get; set; }
  
      // Other properties
  }
  </code></pre>
  
  <p>And the test:</p>
  
  <pre><code>public void PushEventTest()
      {
          const string Address1 = "A:2.1.1";
          const string Address2 = "A:2.1.2";
  
          var eventServiceMock = new Mock&lt;IEventService&gt;();
  
          var scheduler = new TestScheduler();
          var target = new EventManager(eventServiceMock.Object, scheduler);
          var eventMessageA1 = new EventMessage { NetworkAddress = Address1, EventCode = 1, Attribute = 4 };
          var eventMessageB1 = new EventMessage { NetworkAddress = Address2, EventCode = 1, Attribute = 5 };
          var eventMessageA2 = new EventMessage { NetworkAddress = Address1, EventCode = 1, Attribute = 4 };
          scheduler.Schedule(() =&gt; target.PushEvent(eventMessageA1));
          scheduler.Schedule(TimeSpan.FromSeconds(1), () =&gt; target.PushEvent(eventMessageB1));
          scheduler.Schedule(TimeSpan.FromSeconds(2), () =&gt; target.PushEvent(eventMessageA1));
  
          scheduler.AdvanceTo(TimeSpan.FromSeconds(6).Ticks);
  
          eventServiceMock.Verify(s =&gt; s.Add(It.Is&lt;List&lt;Event&gt;&gt;(list =&gt; list.Count == 2)), Times.Once());
  
          scheduler.Schedule(TimeSpan.FromSeconds(3), () =&gt; target.PushEvent(eventMessageB1));
  
          scheduler.AdvanceTo(TimeSpan.FromSeconds(11).Ticks);
  
          eventServiceMock.Verify(s =&gt; s.Add(It.Is&lt;List&lt;Event&gt;&gt;(list =&gt; list.Count == 1)), Times.Once());
      }
  </code></pre>
  
  <p>Additionally, I remark again that it is really important that the software could run for days without problems, handling thousands of messages.
  To make it clear: the test doesn't pass with the current implementation.</p>