Microsoft Reactive Extensions Rx Observer Pattern Kata

Overview

NOTE: So this is not a discussion about Reactive Extensions by Microsoft in relation to what they are and why they should be used. This has already been coverved extensively on Microsoft's website HERE. 

The point of this post is to provide details of a Kata that can be practiced. The design pattern implemented here is the Observer pattern. Instead of implementing this pattern using Events, I am implementing this using Observables.


What does the Application do?
This is a simulation of a user with a phone that has an application that will post messages. There are other users that have phones that will subscribe to these messages.

How does the application work?
The main form is the TextMessageManager. This form has the following operations available:

1. Post a new message
2. Launch a new Phone interface. This button allows you to open multiple Phone windows to simulate having multiple phones subscribing to your messages.


Figure 1: TextMessageManager

When you Open New Phone Window you get a new window that will show the Date/Time the message was posted and the text of the message.

Figure 2: Phone

 

So when you post a new message you get the following

Figure 3: New Message Posted

You can open a new phone and then post a new message and the new phone and the existing phone will both display the new message. You can open as many new phones as you want.

Figure 4: Two phone windows

How does this work?

There are two main classes at work here: 

1. MessageList.cs

2. Phone.cs

MessageList.cs

This class is responsible for publishing the messages to the subscribers. This is responsible for publishing events into the event stream using the OnNext method. The OnNext method is responsible for notifying all subscribed observers that a new message has been published and it is also responbile for sending the observers the value or in the this case, the text of the message.

namespace ReactiveExtensions

{

    public class MessageList

    {

        ISubject<string> newMessage = new Subject<string>();

        public virtual void WhenNewMessage(string message)

        {

            newMessage.OnNext(message);

        }

        public IObservable<string> NewMessage { get { return newMessage; } }

    }

}

The ISubject<string> is setting up a stream of strings and when new Subject<string>() is called this is setting up a new source of events. When the TextMessageManger calls the WhenNewMessage method, the OnNext method is called which sends a new event into the newMessage event stream with a new string value message which is the text of the message that has been posted.

Phone.cs

This class is responsible for subscribing and unsubscribing to the messages that the TextMessageManager is publishing. This class subscribes to the newMessage event stream. When a new message value is sent into the event stream, this will write the value of the event into the window and results in you seeing the Date/Time and Message being displayed on screen.

In order to unsubscribe from the event stream, you need to call the Dispose() method. If you don't do this, when you close the window without notifying the event stream that you no longer want to know about new messages, the OnNext will try and send a new event to something that no longer exists and you will get an error.

namespace ReactiveExtensions

{

    public partial class Phone : Form

    {

        IDisposable newMessageSubscription = new Subject<string>();

        public Phone(MessageList messageSubscription)

        {

            InitializeComponent();

            newMessageSubscription = messageSubscription.NewMessage.Subscribe(text =>

                dataGridMessageList.Rows.Add(new string[] {DateTime.Now.ToString(), text.ToString()}));

        }

        void Phone_FormClosing(object o, FormClosingEventArgs args)

        {

            newMessageSubscription.Dispose();

        }

    }

}

Here is the source code for this kata and if you have any questions or comments feel free to leave them below.