Fluent State Observer for Reactive Extensions / Rx Framework

I've been messing around with the new Reactive Extensions for a couple weeks and wanted to build something that would allow you to match patterns in an Observable stream.

For more detailed information about the Reactive Extensions, check out the Reactive Extensions API In-Depth Intro Video and other In-Depth Reactive Extension videos.

I came up with the following library and put it up on Codeplex as a project called Fluent State Observer in case I ever need to get to it again. I'm going to paste the project home page content here strictly for SEO purposes.

---

Description

The Fluent State Observer helps perform actions based on a pattern in an IObservable "stream".

This library is developed in C# and uses the Reactive (Rx) Extensions library for it's implementation of the IObservable pattern and helper extensions for creating anonymous observers.

For more detailed information about the Reactive Extensions, check out the Reactive Extensions API In-Depth Intro Video and other In-Depth Reactive Extension videos.

This project is developed by Jacob Gable.

Example Uses

The following examples are from the test project. The test project includes a mocked IObservable that pumps out TestElements in succession. The call to obs.ProcessBlocks() begins the stream of Observable items.

See the Observable Series these tests use

Match Complex Patterns

[TestMethod()]

public void Should_Match_FollowedBy_Patterns()
{
// New up with a Mock IObservable that just pumps out items in this pattern ( 1, 2, 2, 3, 1, 2, 3, 4 )
var obs = TestObservables.BasicObservable();
var target = new FluentStateObserver(obs);

bool didMatch = false;
target
.StartsWith(x => x.TypeOfElement == ElementType.TestType1)
.FollowedBy(x => x.TypeOfElement == ElementType.TestType3)
.EndsWith(x => x.TypeOfElement == ElementType.TestType4)
.OnMatch(x => didMatch = true);

obs.ProcessBlocks();

Assert.IsTrue(didMatch);
}

Non-Greedy Matching

Matches only when immediately followed by something.

[TestMethod()]
public void Should_Match_FollowedImmediately_Patterns()
{
var obs = TestObservables.BasicObservable();
var target = new FluentStateObserver(obs);

bool didMatch = false;
target
.StartsWith(x => x.TypeOfElement == ElementType.TestType1)
.FollowedImmediatelyBy(x => x.TypeOfElement == ElementType.TestType2)
.EndsWith(x => x.TypeOfElement == ElementType.TestType4)
.OnMatch(x => didMatch = true);

obs.ProcessBlocks();

Assert.IsTrue(didMatch);
}

Specify Reset / Disqualifying Conditions At Each Transition

Match 1, 5, 4, 3, 4, but not 1, 2, 4, 3, 4

[TestMethod()]
public void Should_Reset_When_A_Reset_Condition_Is_Satisfied()
{
var obs = TestObservables.BasicObservable();
var target = new FluentStateObserver(obs);

int matches = 0;
target
.StartsWith(x => x.TypeOfElement == ElementType.TestType1)
.FollowedBy(x => x.TypeOfElement == ElementType.TestType3)
.ResetOn(x => x.TypeOfElement == ElementType.TestType2)
.EndsWith(x => x.TypeOfElement == ElementType.TestType4)
.OnMatch(x => matches++);

obs.ProcessBlocks();

Assert.AreEqual(0, matches);
}

The Mock Observable in these tests publishes the following series

// Basically, [ 1, 2, 2, 3, 1, 2, 3, 4 ]
new TestElement()
{
Message = "Message 1",
TypeOfElement = ElementType.TestType1
});

new TestElement()
{
Message = "Message 2",
TypeOfElement = ElementType.TestType2
});

new TestElement()
{
Message = "Message 2",
TypeOfElement = ElementType.TestType2
});

new TestElement()
{
Message = "Message 3",
TypeOfElement = ElementType.TestType3
});

new TestElement()
{
Message = "Message 1",
TypeOfElement = ElementType.TestType1
});

new TestElement()
{
Message = "Message 2",
TypeOfElement = ElementType.TestType2
});

new TestElement()
{
Message = "Message 3",
TypeOfElement = ElementType.TestType3
});

new TestElement()
{
Message = "Message 4",
TypeOfElement = ElementType.TestType4
});

That's it. Hope someone finds it useful.

Now Playing: Drake - I'm Going In (Explicit)