101-rx-samples.

md 6/28/2019

101 Rx Samples in C#
From [Link]

This was taken from [Link] because I wanted to be able to read it more
comfortable with syntax highlighting.

Here's the unedited original, translated to Github Markdown glory:

101 Rx Samples - a work in progress

You!

Yes, you, the one who is still scratching their head trying to figure out this Rx thing. As you learn and
explore, please feel free add your own samples here (or tweak existing ones!) Anyone can (and
should!) edit this page. (edit button is at the bottom right of each page)

(and sorry for the years of spam pages there - they should be gone now. Thanks for marking them. -Rob)

Table of Contents
101 Rx Samples in C
101 Rx Samples - a work in progress
Table of Contents
Asynchronous Background Operations
Start - Run Code Asynchronously
Run a method asynchronously on demand
CombineLatest - Parallel Execution
Create With Disposable & Scheduler - Canceling an asynchronous operation
Observation Operators
Observing an Event - Simple
Observing an Event - Simple (expanded)
Observing MouseMove in Silverlight
Observing an Event - Generic
Observing an Event - Non-Generic
Observing an Asynchronous Operation
Observing a Generic IEnumerable
Observing a Non-Generic IEnumerable - Single Type
Observing a Non-Generic IEnumerable - Multiple Types
Observing the Passing of Time
Restriction Operators
Where - Simple
Where - Drilldown
Projection Operators
Select - Simple
Select - Transformation

1 / 35
[Link] 6/28/2019

Select - Indexed
Grouping
Group By - Simple
Time-Related Operators
Buffer - Simple
Delay - Simple
Interval - Simple
Sample - Simple
Throttle - Simple
Interval - With TimeInterval() - Simple
Interval - With TimeInterval() - Remove
Timeout - Simple
Timer - Simple
Timestamp - Simple
Timestamp - Remove
Window and Joins
Window
GroupJoin - Joins two streams matching by one of their attributes
Range
Range - Prints from 1 to 10.
Generate
Generate - simple
ISubject and ISubject<T1, T2>
Ping Pong Actor Model with ISubject<T1, T2>
Combination Operators
Merge
Publish - Sharing a subscription with multiple Observers
Zip
CombineLatest
Concat - cold observable
Concat - hot observable
Make your class native to IObservable
Use Subject as backend for IObservable

Asynchronous Background Operations

Start - Run Code Asynchronously

public static void StartBackgroundWork() {

[Link]("Shows use of Start to start on a background
thread:");
var o = [Link](() =>
{
//This starts on a background thread.
[Link]("From background thread. Does not block main
thread.");
[Link]("Calculating...");

2 / 35
[Link] 6/28/2019

[Link](3000);
[Link]("Background work completed.");
}).Finally(() => [Link]("Main thread completed."));
[Link]("\r\n\t In Main Thread...\r\n");
[Link](); // Wait for completion of background operation.
}

Run a method asynchronously on demand

Execute a long-running method asynchronously. The method does not start running until there is a subscriber.
The method is started every time the observable is created and subscribed, so there could be more than one
running at once.

// Synchronous operation
public DataType DoLongRunningOperation(string param)
{
...
}

public IObservable<DataType> LongRunningOperationAsync(string param)

{
return [Link]<DataType>(
o => [Link]<string,DataType>(DoLongRunningOperation)
(param).Subscribe(o)
);
}

CombineLatest - Parallel Execution

Merges the specified observable sequences into one observable sequence by emitting a list with the latest
source elements whenever any of the observable sequences produces an element.

public async void ParallelExecutionTest()

{
var o = [Link](
[Link](() => { [Link]("Executing 1st on
Thread: {0}", [Link]); return "Result A";
}),
[Link](() => { [Link]("Executing 2nd on
Thread: {0}", [Link]); return "Result B";
}),
[Link](() => { [Link]("Executing 3rd on
Thread: {0}", [Link]); return "Result C"; })
).Finally(() => [Link]("Done!"));

foreach (string r in await [Link]())

[Link](r);
}

3 / 35
[Link] 6/28/2019

Result
Executing 1st on Thread: 3
Executing 2nd on Thread: 4
Executing 3rd on Thread: 3
Done!
Result A
Result B
Result C

Note Was ForkJoin which is no longer supported. CombineLatest gives the same result.)

Create With Disposable & Scheduler - Canceling an asynchronous operation

This sample starts a background operation that generates a sequence of integers until it is canceled by the
main thread. To start the background operation new the Scheduler class is used and a
CancellationTokenSource is indirectly created by a [Link].
Please check out the MSDN documentation on [Link] to learn more
about cancellation source.

IObservable<int> ob =
[Link]<int>(o =>
{
var cancel = new CancellationDisposable(); // internally
creates a new CancellationTokenSource
[Link](() =>
{
int i = 0;
for (; ; )
{
[Link](200); // here we do the long lasting
background operation
if (![Link]) //
check cancel token periodically
[Link](i++);
else
{
[Link]("Aborting because cancel
event was signaled!");
[Link]();
return;
}
}
}
);

return cancel;
}
);

IDisposable subscription = [Link](i => [Link](i));

4 / 35
[Link] 6/28/2019

[Link]("Press any key to cancel");

[Link]();
[Link]();
[Link]("Press any key to quit");
[Link](); // give background thread chance to write the cancel
acknowledge message

Observation Operators
Observing an Event - Simple

class ObserveEvent_Simple
{
public static event EventHandler SimpleEvent;
static void Main()
{
// To consume SimpleEvent as an IObservable:
var eventAsObservable = [Link](
ev => SimpleEvent += ev,
ev => SimpleEvent -= ev);
}
}

Alternately, you can use EventArgs:

public static event EventHandler<EventArgs> SimpleEvent;

private static void Main(string[] args) {

var eventAsObservable = [Link]<EventArgs>
(ev => SimpleEvent += ev,
ev => SimpleEvent -= ev);
}

Observing an Event - Simple (expanded)

class ObserveEvent_Simple
{
public static event EventHandler SimpleEvent;

private static void Main()

{
[Link]("Setup observable");
// To consume SimpleEvent as an IObservable:
var eventAsObservable = [Link](
ev => SimpleEvent += ev,
ev => SimpleEvent -= ev);

5 / 35
[Link] 6/28/2019

// SimpleEvent is null until we subscribe

[Link](SimpleEvent == null ? "SimpleEvent == null" :
"SimpleEvent != null");

[Link]("Subscribe");
//Create two event subscribers
var s = [Link](args =>
[Link]("Received event for s subscriber"));
var t = [Link](args =>
[Link]("Received event for t subscriber"));

// After subscribing the event handler has been added

[Link](SimpleEvent == null ? "SimpleEvent == null" :
"SimpleEvent != null");

[Link]("Raise event");
if (null != SimpleEvent)
{
SimpleEvent(null, [Link]);
}

// Allow some time before unsubscribing or event may not happen

[Link](100);

[Link]("Unsubscribe");
[Link]();
[Link]();

// After unsubscribing the event handler has been removed

[Link](SimpleEvent == null ? "SimpleEvent == null" :
"SimpleEvent != null");

[Link]();
}
}

Observing MouseMove in Silverlight

var mouseMove = [Link]<MouseEventArgs>(this,

"MouseMove");
[Link]()
.Subscribe(args =>
[Link]([Link](this)));

Note that a reference to [Link] is required for ObserveOnDispatcher which is in

Nuget as Reactive Extensions - Silverlight Helpers.

Observing an Event - Generic

6 / 35
[Link] 6/28/2019

class ObserveEvent_Generic
{
public class SomeEventArgs : EventArgs { }
public static event EventHandler<SomeEventArgs> GenericEvent;

static void Main()

{
// To consume GenericEvent as an IObservable:
IObservable<EventPattern<SomeEventArgs>> eventAsObservable =
[Link]<SomeEventArgs>(
ev => GenericEvent += ev,
ev => GenericEvent -= ev );
}
}

Observing an Event - Non-Generic

class ObserveEvent_NonGeneric
{
public class SomeEventArgs : EventArgs { }
public delegate void SomeNonGenericEventHandler(object sender,
SomeEventArgs e);
public static event SomeNonGenericEventHandler NonGenericEvent;

static void Main()

{
// To consume NonGenericEvent as an IObservable, first inspect the
type of EventArgs used in the second parameter of the delegate.
// In this case, it is SomeEventArgs. Then, use as shown below.
IObservable<IEvent<SomeEventArgs>> eventAsObservable =
[Link](
(EventHandler<SomeEventArgs> ev) => new
SomeNonGenericEventHandler(ev),
ev => NonGenericEvent += ev,
ev => NonGenericEvent -= ev);
}
}

Observing an Asynchronous Operation

class Observe_IAsync
{
static void Main()
{
// We will use Stream's BeginRead and EndRead for this sample.
Stream inputStream = [Link]();

// To convert an asynchronous operation that uses the IAsyncResult

7 / 35
[Link] 6/28/2019

pattern to a function that returns an IObservable, use the following

format.
// For the generic arguments, specify the types of the arguments
of the Begin* method, up to the AsyncCallback.
// If the End* method returns a value, append this as your final
generic argument.
var read = [Link]<byte[], int, int, int>
([Link], [Link]);

// Now, you can get an IObservable instead of an IAsyncResult when

calling it.
byte[] someBytes = new byte[10];
IObservable<int> observable = read(someBytes, 0, 10);
}
}

Be aware that while the code above formally provides an observable, this is not enough for most intended
uses. For more information, see
Creating an observable sequence and
c# - What is the proper way to create an Observable which reads a stream to the end - Stack Overflow.

Observing a Generic IEnumerable

class Observe_GenericIEnumerable
{
static void Main()
{
IEnumerable<int> someInts = new List<int> { 1, 2, 3, 4, 5 };

// To convert a generic IEnumerable into an IObservable, use the

ToObservable extension method.
IObservable<int> observable = [Link]();
}
}

Observing a Non-Generic IEnumerable - Single Type

class Observe_NonGenericIEnumerableSingleType
{
static void Main()
{
IEnumerable someInts = new object[] { 1, 2, 3, 4, 5 };

// To convert a non-generic IEnumerable that contains elements of

a single type,
// first use Cast<> to change the non-generic enumerable into a
generic enumerable,
// then use ToObservable.

8 / 35
[Link] 6/28/2019

IObservable<int> observable = [Link]<int>().ToObservable();

}
}

Observing a Non-Generic IEnumerable - Multiple Types

Observing the Passing of Time

class Observe_Time
{
static void Main()
{
// To observe time passing, use the [Link] function.
// It will notify you on a time interval you specify.

// 0 after 1s, 1 after 2s, 2 after 3s, etc.

IObservable<long> oneNumberPerSecond =
[Link]([Link](1));
IObservable<long> alsoOneNumberPerSecond =
[Link](1000 /* milliseconds */);
}
}

Restriction Operators
Where - Simple

class Where_Simple
{
static void Main()
{
var oneNumberPerSecond =
[Link]([Link](1));

var lowNums = from n in oneNumberPerSecond

where n < 5
select n;

[Link]("Numbers < 5:");

[Link](lowNum =>
{
[Link](lowNum);
});

[Link]();
}
}

9 / 35
[Link] 6/28/2019

Result
Numbers < 5:
0 (after 1s)
1 (after 2s)
2 (after 3s)
3 (after 4s)
4 (after 5s)

Where - Drilldown

class Where_DrillDown
{
class Customer
{
public Customer() { Orders = new ObservableCollection<Order>(); }
public string CustomerName { get; set; }
public string Region { get; set; }
public ObservableCollection<Order> Orders { get; private set; }
}

class Order
{
public int OrderId { get; set; }
public DateTimeOffset OrderDate { get; set; }
}

static void Main()

{
var customers = new ObservableCollection<Customer>();

var customerChanges = [Link](

(EventHandler<NotifyCollectionChangedEventArgs> ev)
=> new NotifyCollectionChangedEventHandler(ev),
ev => [Link] += ev,
ev => [Link] -= ev);

var watchForNewCustomersFromWashington =
from c in customerChanges
where [Link] == [Link]
from cus in [Link]<Customer>
().ToObservable()
where [Link] == "WA"
select cus;

[Link]("New customers from Washington and their

orders:");

[Link](cus =>
{
[Link]("Customer {0}:", [Link]);

10 / 35
[Link] 6/28/2019

foreach (var order in [Link])

{
[Link]("Order {0}: {1}", [Link],
[Link]);
}
});

[Link](new Customer
{
CustomerName = "Lazy K Kountry Store",
Region = "WA",
Orders = { new Order { OrderDate = [Link], OrderId
= 1 } }
});

[Link](1000);
[Link](new Customer
{
CustomerName = "Joe's Food Shop",
Region = "NY",
Orders = { new Order { OrderDate = [Link], OrderId
= 2 } }
});

[Link](1000);
[Link](new Customer
{
CustomerName = "Trail's Head Gourmet Provisioners",
Region = "WA",
Orders = { new Order { OrderDate = [Link], OrderId
= 3 } }
});

[Link]();
}
}

Result
New customers from Washington and their orders:
Customer Lazy K Kountry Store: (after 0s)
Order 1: 11/20/2009 11:52:02 AM -06:00
Customer Trail's Head Gourmet Provisioners: (after 2s)
Order 3: 11/20/2009 11:52:04 AM -06:00

Projection Operators
Select - Simple

class Select_Simple
{

11 / 35
[Link] 6/28/2019

static void Main()

{
var oneNumberPerSecond =
[Link]([Link](1));

var numbersTimesTwo = from n in oneNumberPerSecond

select n * 2;

[Link]("Numbers * 2:");

[Link](num =>
{
[Link](num);
});

[Link]();
}
}

Result
Numbers * 2:
0 (after 1s)
2 (after 2s)
4 (after 3s)
6 (after 4s)
8 (after 5s)

Select - Transformation

class Select_Transform
{
static void Main()
{
var oneNumberPerSecond =
[Link]([Link](1));

var stringsFromNumbers = from n in oneNumberPerSecond

select new string('*', (int)n);

[Link]("Strings from numbers:");

[Link](num =>
{
[Link](num);
});

[Link]();
}
}

12 / 35
[Link] 6/28/2019

Result
Strings from numbers:
(after 0s)
(after 1s)
(after 2s)
(after 3s)
*** (after 4s)
(after 5s)
(after 6s)

Select - Indexed

class Where_Indexed
{
class TimeIndex
{
public TimeIndex(int index, DateTimeOffset time)
{
Index = index;
Time = time;
}
public int Index { get; set; }
public DateTimeOffset Time { get; set; }
}

static void Main()

{
var clock = [Link]([Link](1))
.Select((t, index) => new TimeIndex(index,
[Link]));

[Link](timeIndex =>
{
[Link](
"Ding dong. The time is now {0:T}. This is event number
{1}.",
[Link],
[Link]);
});

[Link]();
}
}

Result
Ding dong. The time is now 1:55:00 PM. This is event number 0. (after 0s)
Ding dong. The time is now 1:55:01 PM. This is event number 1. (after 1s)
Ding dong. The time is now 1:55:02 PM. This is event number 2. (after 2s)
Ding dong. The time is now 1:55:03 PM. This is event number 3. (after 3s)

13 / 35
[Link] 6/28/2019

Ding dong. The time is now 1:55:04 PM. This is event number 4. (after 4s)
Ding dong. The time is now 1:55:05 PM. This is event number 5. (after 5s)

Grouping
Group By - Simple

This example counts how many time you press each key as you furiously hit the keyboard. :)

class GroupBy_Simple
{
static IEnumerable<ConsoleKeyInfo> KeyPresses()
{
for (; ; )
{
var currentKey = [Link](true);

if ([Link] == [Link])
yield break;
else
yield return currentKey;
}
}
static void Main()
{
var timeToStop = new ManualResetEvent(false);
var keyPresses = KeyPresses().ToObservable();

var groupedKeyPresses =
from k in keyPresses
group k by [Link] into keyPressGroup
select keyPressGroup;

[Link]("Press Enter to stop. Now bang that

keyboard!");

[Link](keyPressGroup =>
{
int numberPresses = 0;

[Link](keyPress =>
{
[Link](
"You pressed the {0} key {1} time(s)!",
[Link],
++numberPresses);
},
() => [Link]());
});

[Link]();

14 / 35
[Link] 6/28/2019

}
}

Result
Depends on what you press! But something like:
Press Enter to stop. Now bang that keyboard!
You pressed the A key 1 time(s)!
You pressed the A key 2 time(s)!
You pressed the B key 1 time(s)!
You pressed the B key 2 time(s)!
You pressed the C key 1 time(s)!
You pressed the C key 2 time(s)!
You pressed the C key 3 time(s)!
You pressed the A key 3 time(s)!
You pressed the B key 3 time(s)!
You pressed the A key 4 time(s)!
You pressed the A key 5 time(s)!
You pressed the C key 4 time(s)!

Time-Related Operators
Buffer - Simple

Buffer has a strange name, but a simple concept.

Imagine an email program that checks for new mail every 5 minutes. While you can receive mail at any instant
in time, you only get a batch of emails at every five minute mark.
Let's use Buffer to simulate this.

class Buffer_Simple
{
static IEnumerable<string> EndlessBarrageOfEmail()
{
var random = new Random();
var emails = new List<String> { "Here is an email!", "Another
email!", "Yet another email!" };
for (; ; )
{
// Return some random emails at random intervals.
yield return emails[[Link]([Link])];
[Link]([Link](1000));
}
}
static void Main()
{
var myInbox = EndlessBarrageOfEmail().ToObservable();

// Instead of making you wait 5 minutes, we will just check every

three seconds instead. :)
var getMailEveryThreeSeconds =
15 / 35
[Link] 6/28/2019

[Link]([Link](3)); // Was .BufferWithTime(...

[Link](emails =>
{
[Link]("You've got {0} new messages! Here they
are!", [Link]());
foreach (var email in emails)
{
[Link]("> {0}", email);
}
[Link]();
});

[Link]();
}
}

Result
You've got 5 new messages! Here they are! (after 3s)
> Here is an email!
> Another email!
> Here is an email!
> Another email!
> Here is an email!

You've got 6 new messages! Here they are! (after 6s)

> Another email!
> Another email!
> Here is an email!
> Here is an email!
> Another email!
> Another email!

Delay - Simple

class Delay_Simple
{
static void Main()
{
var oneNumberEveryFiveSeconds =
[Link]([Link](5));

// Instant echo
[Link](num =>
{
[Link](num);
});

// One second delay

16 / 35
[Link] 6/28/2019

[Link]([Link](1)).Subscribe(num =>
{
[Link]("...{0}...", num);
});

// Two second delay

[Link]([Link](2)).Subscribe(num =>
{
[Link]("......{0}......", num);
});

[Link]();
}
}

Result
0 (after 5s)
…0… (after 6s)
……0…… (after 7s)
1 (after 10s)
…1… (after 11s)
……1…… (after 12s)

Interval - Simple

internal class Interval_Simple

{
private static void Main()
{
IObservable<long> observable =
[Link]([Link](1));

using ([Link]([Link]))
{
[Link]("Press any key to unsubscribe");
[Link]();
}

[Link]("Press any key to exit");

[Link]();
}
}

Result
0 (after 1s)
1 (after 2s)
2 (after 3s)

17 / 35
[Link] 6/28/2019

3 (after 4s)
…

Sample - Simple

internal class Sample_Simple

{
private static void Main()
{
// Generate sequence of numbers, (an interval of 50 ms seems to
result in approx 16 per second).
IObservable<long> observable =
[Link]([Link](50));

// Sample the sequence every second

using
([Link]([Link](1)).Timestamp().Subscribe(
x => [Link]("{0}: {1}", [Link], [Link])))
{
[Link]("Press any key to unsubscribe");
[Link]();
}

[Link]("Press any key to exit");

[Link]();
}
}

Result
15: 24/11/2009 15:40:45 (after 1s)
31: 24/11/2009 15:40:46 (after 2s)
47: 24/11/2009 15:40:47 (after 3s)
64: 24/11/2009 15:40:48 (after 4s)
…

Throttle - Simple

Throttle stops the flow of events until no more events are produced for a specified period of time. For
example, if you throttle a TextChanged event of a textbox to .5 seconds, no events will be passed until the user
has stopped typing for .5 seconds. This is useful in search boxes where you do not want to start a new search
after every keystroke, but want to wait until the user pauses.

SearchTextChangedObservable =
[Link]<TextChangedEventArgs>([Link],
"TextChanged");
_currentSubscription =
[Link]([Link](.5)).ObserveOnDi
spatcher().Subscribe(e => [Link]([Link]));

18 / 35
[Link] 6/28/2019

Here is another example:

internal class Throttle_Simple

{
// Generates events with interval that alternates between 500ms and
1000ms every 5 events
static IEnumerable<int> GenerateAlternatingFastAndSlowEvents()
{
int i = 0;

while(true)
{
if(i > 1000)
{
yield break;
}
yield return i;
[Link]( i++ % 10 < 5 ? 500 : 1000);
}
}

private static void Main()

{
var observable =
GenerateAlternatingFastAndSlowEvents().ToObservable().Timestamp();
var throttled =
[Link]([Link](750));

using ([Link](x => [Link]("{0}: {1}",

[Link], [Link])))
{
[Link]("Press any key to unsubscribe");
[Link]();
}

[Link]("Press any key to exit");

[Link]();
}
}

Result
5:
6:
7:
8:
9:
15:
16:
17:

19 / 35
[Link] 6/28/2019

18:
19:
…etc

Interval - With TimeInterval() - Simple

internal class TimeInterval_Simple

{
// Like TimeStamp but gives the time-interval between successive
values
private static void Main()
{
var observable =
[Link]([Link](750)).TimeInterval();

using ([Link](
x => [Link]("{0}: {1}", [Link], [Link])))
{
[Link]("Press any key to unsubscribe");
[Link]();
}

[Link]("Press any key to exit");

[Link]();
}
}

Result
0: 00:00:00.8090459 (1st value)
1: 00:00:00.7610435 (2nd value)
2: 00:00:00.7650438 (3rd value)
…

Interval - With TimeInterval() - Remove

internal class TimeInterval_Remove

{
private static void Main()
{
// Add a time interval
var observable =
[Link]([Link](750)).TimeInterval();

// Remove it again
using
([Link]().Subscribe([Link]))
{
[Link]("Press any key to unsubscribe");
[Link]();

20 / 35
[Link] 6/28/2019

[Link]("Press any key to exit");

[Link]();
}
}

Result
0
1
2
…

Timeout - Simple

internal class Timeout_Simple

{
private static void Main()
{
[Link]([Link]);

// create a single event in 10 seconds time

var observable =
[Link]([Link](10)).Timestamp();

// raise exception if no event received within 9 seconds

var observableWithTimeout = [Link](observable,
[Link](9));

using ([Link](
x => [Link]("{0}: {1}", [Link], [Link]),
ex => [Link]("{0} {1}", [Link], [Link])))
{
[Link]("Press any key to unsubscribe");
[Link]();
}

[Link]("Press any key to exit");

[Link]();
}
}

Result
02/12/2009 10:13:00
Press any key to unsubscribe
The operation has timed out. 02/12/2009 10:13:09
…

Timer - Simple
21 / 35
[Link] 6/28/2019

[Link] is a simple wrapper around [Link].

internal class Timer_Simple

{
private static void Main()
{
[Link]([Link]);

var observable = [Link]([Link](5),

[Link](1)).Timestamp();

// or, equivalently
// var observable = [Link]([Link] +
[Link](5),
//
[Link](1)).Timestamp();

using ([Link](
x => [Link]("{0}: {1}", [Link], [Link])))
{
[Link]("Press any key to unsubscribe");
[Link]();
}

[Link]("Press any key to exit");

[Link]();
}
}

Result
02/12/2009 10:02:29
Press any key to unsubscribe
0: 02/12/2009 10:02:34(after 5s)
1: 02/12/2009 10:02:35 (after 6s)
2: 02/12/2009 10:02:36 (after 7s)
…

Timestamp - Simple

Adds a TimeStamp to each element using the system's local time.

internal class Timestamp_Simple

{
private static void Main()
{
var observable =
[Link]([Link](1)).Timestamp();

22 / 35
[Link] 6/28/2019

using ([Link](
x => [Link]("{0}: {1}", [Link], [Link])))
{
[Link]("Press any key to unsubscribe");
[Link]();
}

[Link]("Press any key to exit");

[Link]();
}
}

Result
0: 24/11/2009 15:40:45 (after 1s)
1: 24/11/2009 15:40:46 (after 2s)
2: 24/11/2009 15:40:47 (after 3s)
3: 24/11/2009 15:40:48 (after 4s)
…

Timestamp - Remove

internal class Timestamp_Remove

{
private static void Main()
{
// Add timestamp
var observable =
[Link]([Link](1)).Timestamp();

// Remove it
using ([Link]().Subscribe([Link]))
{
[Link]("Press any key to unsubscribe");
[Link]();
}

[Link]("Press any key to exit");

[Link]();
}
}

Result
0 (after 1s)
1 (after 2s)
2 (after 3s)
3 (after 4s)
…

Window and Joins

23 / 35
[Link] 6/28/2019

Window

Divides a stream into "Windows" of time. For example, 5 five second window would contain all elements
pushed in that five second interval.

IObservable<long> mainSequence =
[Link]([Link](1));
IObservable<IObservable<long>> seqWindowed = [Link](() =>
{
IObservable<long> seqWindowControl =
[Link]([Link](6));
return seqWindowControl;
});

[Link](seqWindow =>
{
[Link]("\nA new window into the main sequence has
opened: {0}\n",
[Link]());
[Link](x => { [Link]("Integer : {0}", x);
});
});

[Link]();

GroupJoin - Joins two streams matching by one of their attributes

var leftList = new List<string[]>();

[Link](new string[] { "2013-01-01 02:00:00", "Batch1" });
[Link](new string[] { "2013-01-01 03:00:00", "Batch2" });
[Link](new string[] { "2013-01-01 04:00:00", "Batch3" });

var rightList = new List<string[]>();

[Link](new string[] { "2013-01-01 01:00:00", "Production=2" });
[Link](new string[] { "2013-01-01 02:00:00", "Production=0" });
[Link](new string[] { "2013-01-01 03:00:00", "Production=3" });

var l = [Link]();
var r = [Link]();

var q = [Link](r,
_ => [Link]<Unit>(), // windows from each left event going
on forever
_ => [Link]<Unit>(), // windows from each right event going
on forever
(left, obsOfRight) => [Link](left, obsOfRight)); // create tuple
of left event with observable of right events

// e is a tuple with two items, left and obsOfRight

[Link](e =>
24 / 35
[Link] 6/28/2019

{
var xs = e.Item2;
[Link](
x => x[0] == e.Item1[0]) // filter only when datetime matches
.Subscribe(
v =>
{
[Link](
[Link]("{0},{1} and {2},{3} occur at the same time",
e.Item1[0],
e.Item1[1],
v[0],
v[1]
));
});
});

Range
Generates a Range of values. Useful for testing purposes.

Range - Prints from 1 to 10.

IObservable<int> source = [Link](1, 10);

IDisposable subscription = [Link](
x => [Link]("OnNext: {0}", x),
ex => [Link]("OnError: {0}", [Link]),
() => [Link]("OnCompleted"));
[Link]("Press ENTER to unsubscribe...");
[Link]();
[Link]();

Generate
There are several overloads for Generate.

Generate - simple

A simple use is to replicate Interval but have the sequence stop.

internal class Generate_Simple

{
private static void Main()
{
var observable =
[Link](1, x => x < 6, x => x + 1, x => x,

x=>[Link](1)).Timestamp();

25 / 35
[Link] 6/28/2019

using ([Link](x => [Link]("{0}, {1}",

[Link], [Link])))
{
[Link]("Press any key to unsubscribe");
[Link]();
}

[Link]("Press any key to exit");

[Link]();
}
}

Result
1: 24/11/2009 15:40:45 (after 1s)
2: 24/11/2009 15:40:46 (after 2s)
3: 24/11/2009 15:40:47 (after 3s)
4: 24/11/2009 15:40:48 (after 4s)
5: 24/11/2009 15:40:49 (after 5s)

ISubject and ISubject<T1, T2>

There are several implementations for ISubject.

Ping Pong Actor Model with ISubject<T1, T2>

using System;
using [Link];
using [Link];

namespace RxPingPong
{
/// <summary>Simple Ping Pong Actor model using Rx </summary>
/// <remarks>
/// You'll need to install the Reactive Extensions (Rx) for this to
work.
/// You can get the installer from <see
href="[Link]
/// </remarks>
class Program
{
static void Main(string[] args)
{
var ping = new Ping();
var pong = new Pong();

[Link]("Press any key to stop ...");

var pongSubscription = [Link](pong);

var pingSubscription = [Link](ping);

26 / 35
[Link] 6/28/2019

[Link]();

[Link]();
[Link]();

[Link]("Ping Pong has completed.");

}
}

class Ping : ISubject<Pong, Ping>

{
#region Implementation of IObserver<Pong>

/// <summary>
/// Notifies the observer of a new value in the sequence.
/// </summary>
public void OnNext(Pong value)
{
[Link]("Ping received Pong.");
}

/// <summary>
/// Notifies the observer that an exception has occurred.
/// </summary>
public void OnError(Exception exception)
{
[Link]("Ping experienced an exception and had to
quit playing.");
}

/// <summary>
/// Notifies the observer of the end of the sequence.
/// </summary>
public void OnCompleted()
{
[Link]("Ping finished.");
}

#endregion

#region Implementation of IObservable<Ping>

/// <summary>
/// Subscribes an observer to the observable sequence.
/// </summary>
public IDisposable Subscribe(IObserver<Ping> observer)
{
return [Link]([Link](2))
.Where(n => n < 10)
.Select(n => this)
.Subscribe(observer);
}

#endregion
27 / 35
[Link] 6/28/2019

#region Implementation of IDisposable

/// <summary>
/// Performs application-defined tasks associated with freeing,
releasing, or resetting unmanaged resources.
/// </summary>
/// <filterpriority>2</filterpriority>
public void Dispose()
{
OnCompleted();
}

#endregion
}

class Pong : ISubject<Ping, Pong>

{
#region Implementation of IObserver<Ping>

/// <summary>
/// Notifies the observer of a new value in the sequence.
/// </summary>
public void OnNext(Ping value)
{
[Link]("Pong received Ping.");
}

/// <summary>
/// Notifies the observer that an exception has occurred.
/// </summary>
public void OnError(Exception exception)
{
[Link]("Pong experienced an exception and had to
quit playing.");
}

/// <summary>
/// Notifies the observer of the end of the sequence.
/// </summary>
public void OnCompleted()
{
[Link]("Pong finished.");
}

#endregion

#region Implementation of IObservable<Pong>

/// <summary>
/// Subscribes an observer to the observable sequence.
/// </summary>
public IDisposable Subscribe(IObserver<Pong> observer)
{
28 / 35
[Link] 6/28/2019

return [Link]([Link](1.5))
.Where(n => n < 10)
.Select(n => this)
.Subscribe(observer);
}

#endregion

#region Implementation of IDisposable

/// <summary>
/// Performs application-defined tasks associated with freeing,
releasing, or resetting unmanaged resources.
/// </summary>
/// <filterpriority>2</filterpriority>
public void Dispose()
{
OnCompleted();
}

#endregion
}
}

Result
1: Ping received Pong.
2: Pong received Ping.
3: Ping received Pong.
4: Pong received Ping.
5: Ping received Pong.

Combination Operators
Merge

The Merge operator combine two or more sequences. In the following example, the two streams are merged
into one so that both are printed with one subscription. Also note the use of "using" to wrap the Observable,
thus ensuring the subscription is Disposed.

class Merge
{
private static IObservable<int> Xs
{
get { return Generate(0, new List<int> {1, 2, 2, 2, 2}); }
}

private static IObservable<int> Ys

{
get { return Generate(100, new List<int> {2, 2, 2, 2, 2}); }
}

29 / 35
[Link] 6/28/2019

private static IObservable<int> Generate(int initialValue, IList<int>

intervals)
{
// work-around for [Link] calling timeInterval before
resultSelector
[Link](0);

return [Link](initialValue,
x => x < initialValue + [Link]
- 1,
x => x + 1,
x => x,
x => [Link](intervals[x -
initialValue]));
}

private static void Main()

{
[Link]("Press any key to unsubscribe");

using ([Link](Ys).Timestamp().Subscribe(
z => [Link]("{0,3}: {1}", [Link], [Link]),
() => [Link]("Completed, press a key")))
{
[Link]();
}

[Link]("Press any key to exit");

[Link]();
}
}

result
0: 11/12/2009 12:17:44
100: 11/12/2009 12:17:45
1: 11/12/2009 12:17:46
101: 11/12/2009 12:17:47
2: 11/12/2009 12:17:48
102: 11/12/2009 12:17:49
3: 11/12/2009 12:17:50
103: 11/12/2009 12:17:51
4: 11/12/2009 12:17:52
104: 11/12/2009 12:17:53

Publish - Sharing a subscription with multiple Observers

class Publish
{
private static void Main()
30 / 35
[Link] 6/28/2019

{
var unshared = [Link](1, 4);

// Each subscription starts a new sequence

[Link](i => [Link]("Unshared Subscription
#1: " + i));
[Link](i => [Link]("Unshared Subscription
#2: " + i));

[Link]();

// By using publish the subscriptions are shared, but the sequence

doesn't start until Connect() is called.
var shared = [Link]();
[Link](i => [Link]("Shared Subscription #1: "
+ i));
[Link](i => [Link]("Shared Subscription #2: "
+ i));
[Link]();

[Link]("Press any key to exit");

[Link]();
}
}

result
Unshared Subscription #1: 1
Unshared Subscription #1: 2
Unshared Subscription #1: 3
Unshared Subscription #1: 4
Unshared Subscription #2: 1
Unshared Subscription #2: 2
Unshared Subscription #2: 3
Unshared Subscription #2: 4

Shared Subscription #1: 1

Shared Subscription #2: 1
Shared Subscription #1: 2
Shared Subscription #2: 2
Shared Subscription #1: 3
Shared Subscription #2: 3
Shared Subscription #1: 4
Shared Subscription #2: 4

Zip

class Zip
{
// same code as above for Merge...

31 / 35
[Link] 6/28/2019

private static void Main()

{
[Link]("Press any key to unsubscribe");

using ([Link](Ys, (x, y) => x + y).Timestamp().Subscribe(

z => [Link]("{0,3}: {1}", [Link], [Link]),
() => [Link]("Completed, press a key")))
{
[Link]();
}

[Link]("Press any key to exit");

[Link]();
}
}

result
100: 11/12/2009 12:17:45
102: 11/12/2009 12:17:47
104: 11/12/2009 12:17:49
106: 11/12/2009 12:17:51
108: 11/12/2009 12:17:53

CombineLatest

class CombineLatest
{
// same code as above for Merge...

private static void Main()

{
[Link]("Press any key to unsubscribe");

using ([Link](Ys, (x, y) => x +

y).Timestamp().Subscribe(
z => [Link]("{0,3}: {1}", [Link], [Link]),
() => [Link]("Completed, press a key")))
{
[Link]();
}

[Link]("Press any key to exit");

[Link]();
}
}

result
100: 11/12/2009 12:17:45

32 / 35
[Link] 6/28/2019

101: 11/12/2009 12:17:46
102: 11/12/2009 12:17:47
103: 11/12/2009 12:17:48
104: 11/12/2009 12:17:49
105: 11/12/2009 12:17:50
106: 11/12/2009 12:17:51
107: 11/12/2009 12:17:52
108: 11/12/2009 12:17:53

Concat - cold observable

class ConcatCold
{
private static IObservable<int> Xs
{
get { return Generate(0, new List<int> {0, 1, 1}); }
}

private static IObservable<int> Ys

{
get { return Generate(100, new List<int> {1, 1, 1}); }
}

// same Generate() method as above for Merge...

private static void Main()

{
[Link]("Press any key to unsubscribe");

[Link]([Link]);

using ([Link](Ys).Timestamp().Subscribe(
z => [Link]("{0,3}: {1}", [Link], [Link]),
() => [Link]("Completed, press a key")))
{
[Link]();
}

[Link]("Press any key to exit");

[Link]();
}
}

result
0: 11/12/2009 12:17:45
1: 11/12/2009 12:17:46
2: 11/12/2009 12:17:47
100: 11/12/2009 12:17:48

33 / 35
[Link] 6/28/2019

101: 11/12/2009 12:17:49
102: 11/12/2009 12:17:50

Concat - hot observable

class ConcatHot
{
private static IObservable<int> Xs
{
get { return Generate(0, new List<int> {0, 1, 1}); }
}

private static IObservable<int> Ys

{
get { return Generate(100, new List<int> {1, 1, 1}).Publish(); }
}

// same Generate() method as above for Merge...

private static void Main()

{
[Link]("Press any key to unsubscribe");

[Link]([Link]);

using ([Link](Ys).Timestamp().Subscribe(
z => [Link]("{0,3}: {1}", [Link], [Link]),
() => [Link]("Completed, press a key")))
{
[Link]();
}

[Link]("Press any key to exit");

[Link]();
}
}

result
0: 11/12/2009 12:17:45
1: 11/12/2009 12:17:46
2: 11/12/2009 12:17:47
102: 11/12/2009 12:17:48

Make your class native to IObservable

If you are about to build new system, you could consider using just IObservable.

Use Subject as backend for IObservable

34 / 35
[Link] 6/28/2019

class UseSubject
{
public class Order
{
private DateTime? _paidDate;

private readonly Subject<Order> _paidSubj = new Subject<Order>();

public IObservable<Order> Paid { get { return
_paidSubj.AsObservable(); } }

public void MarkPaid(DateTime paidDate)

{
_paidDate = paidDate;
_paidSubj.OnNext(this); // Raise PAID event
}
}

private static void Main()

{
var order = new Order();
[Link](_ => [Link]("Paid")); // Subscribe

[Link]([Link]);
}
}

35 / 35