To help you understand what is really happening inside a reactive program, we will write some toy programs to put things in proper context. From a software design point of view, if you keep concurrency/parallelism aside to focus on software interfaces, a reactive Program should have:
- An event source that implements
IObservable<T> - An event sink that implements
IObserver<T> - A mechanism to add subscribers to an event source
- When data appears at the source, subscribers will be notified
Note
In this particular chapter, we have written code using classic C++ constructs. This is because we have not yet introduced Modern C++ constructs. We have also used raw pointers, something which we can mostly avoid while writing Modern C++ code. The code in this chapter is written to conform to the ReactiveX documentation in general. In C++, we do not use inheritance-based techniques like we do in Java or C#.
To kickstart, let us define Observer, Observable, and a CustomException class:
#pragma once
//Common2.h
struct CustomException /*:*public std::exception */ { const char * what() const throw () {
return "C++ Exception";
}
}; The CustomException class is just a placeholder to make the interface complete. Since we have decided that we will only use classic C++ in this chapter, we are not deviating from the std::exception class:
template<class T> class IEnumerator {
public:
virtual bool HasMore() = 0;
virtual T next() = 0;
//--------- Omitted Virtual destructor for brevity
};
template <class T> class IEnumerable{
public:
virtual IEnumerator<T> *GetEnumerator() = 0;
//---------- Omitted Virtual destructor for brevity
};The Enumerable interface is used by the data source from which we can enumerate data and IEnuerator<T> will be used for iteration by the client.
Note
The purpose of defining interfaces for Iterator (IEnuerable<T>/IEnumerator<T>) is to make the reader understand that they are very closely related to the Observer<T>/Observable<T> pattern. We will define Observer<T>/Observable<T> as follows:
template<class T> class IObserver
{
public:
virtual void OnCompleted() = 0;
virtual void OnError(CustomException *exception) = 0;
virtual void OnNext(T value) = 0;
};
template<typename T>
class IObservable
{
public:
virtual bool Subscribe(IObserver<T>& observer) = 0;
};IObserver<T> is the interface that the data sink will use to receive notifications from the data source. The data source will implement the IObservable<T> interface.
Note
We have defined the IObserver<T> interface and it has got three methods. They are OnNext (when the item is notified to the Observer), OnCompleted (when there is no more data), and OnError (when an exception is encountered). Observable<T> is implemented by the event source and event sinks can insert objects that implement IObserver<T> to receive notifications.