C++ Reactive Programming

Event-driven programming is a programming model where flow control is determined by events. Examples of events are mouse clicks, key presses, gestures, sensor data, messages from other programs, and so on. An event-driven application has the mechanism to detect events on a near real-time basis, and respond or react to them by invoking the appropriate event handling procedure. Since the bulk of the earlier event processing programs were written using C/C++, they resorted to low-level techniques such as callbacks (using function pointers) to write those event handlers. Later systems such as Visual Basic, Delphi, and other rapid application development tools did add native support for event-driven programming. To make matters more clear, we will take a tour of the event handling mechanism of the various platforms. This will help readers appreciate the issues that reactive programming models are solving (from a GUI programming context).

#include <X11/Xlib.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

int main(void)
{
    Display *display;
    Window window;
    XEvent event;
    char *msg = "Hello, World!";
    int s;

/* open connection with the server */
    display = XOpenDisplay(NULL);
    if (display == NULL){
        fprintf(stderr, "Cannot open display\n");
        exit(1);
    }
    s = DefaultScreen(display);
    /* create window */
    window = XCreateSimpleWindow(display,
             RootWindow(display, s), 10, 10, 200, 200, 1,
             BlackPixel(display, s), WhitePixel(display, s));

    /* select kind of events we are interested in */
    XSelectInput(display, window, ExposureMask | KeyPressMask);

    /* map (show) the window */
    XMapWindow(display, window);

    /* event loop */
    for (;;)
    {
        XNextEvent(display, &event);

        /* draw or redraw the window */
        if (event.type == Expose)
        {
            XFillRectangle(display, window,
                DefaultGC(display, s), 20, 20, 10, 10);
            XDrawString(display, window,
                DefaultGC(display, s), 50, 50, msg, strlen(msg));
        }
        /* exit on key press */
        if (event.type == KeyPress)
        break;
    }

    /* close connection to server */
    XCloseDisplay(display);

    return 0;
    }

#include <windows.h>
//----- Prtotype for the Event Handler Function
LRESULT CALLBACK WndProc(HWND hWnd, UINT message,
                         WPARAM wParam, LPARAM lParam);
//--------------- Entry point for a Idiomatic Windows API function
int WINAPI WinMain(HINSTANCE hInstance,
              HINSTANCE hPrevInstance, LPSTR lpCmdLine, int nCmdShow)
{

MSG msg = {0};
WNDCLASS wc = {0};
wc.lpfnWndProc = WndProc;
wc.hInstance = hInstance;
wc.hbrBackground = (HBRUSH)(COLOR_BACKGROUND);
wc.lpszClassName = "minwindowsapp";
if( !RegisterClass(&wc) )
  return 1;

if( !CreateWindow(wc.lpszClassName,
                  "Minimal Windows Application",
                  WS_OVERLAPPEDWINDOW|WS_VISIBLE,
                  0,0,640,480,0,0,hInstance,NULL))
    return 2;

    while( GetMessage( &msg, NULL, 0, 0 ) > 0 )
        DispatchMessage( &msg );
    return 0;
}

LRESULT CALLBACK WndProc(HWND hWnd, UINT message,
                         WPARAM wParam, LPARAM lParam) {
switch(message){
  case WM_CLOSE:
    PostQuitMessage(0);break;
  default:
    return DefWindowProc(hWnd, message, wParam, lParam);
}
return 0;
}

#include <qapplication.h>
#include <qdialog.h>
#include <qmessagebox.h>
#include <qobject.h>
#include <qpushbutton.h>

class MyApp : public QDialog {
  Q_OBJECT
public:
    MyApp(QObject* /*parent*/ = 0):
    button(this)
    {
      button.setText("Hello world!"); button.resize(100, 30);

      // When the button is clicked, run button_clicked
      connect(&button,
              &QPushButton::clicked, this, &MyApp::button_clicked);
    }

public slots:
    void button_clicked() {
      QMessageBox box;
      box.setWindowTitle("Howdy");
      box.setText("You clicked the button");
      box.show();
      box.exec();
    }

protected:
  QPushButton button;
};

int main(int argc, char** argv) {
  QApplication app(argc, argv);
  MyApp myapp;
  myapp.show();
  return app.exec();
}

BEGIN_MESSAGE_MAP(CClockFrame,CFrameWnd)
    ON_WM_CREATE()
    ON_WM_PAINT()
    ON_WM_TIMER()
END_MESSAGE_MAP()

Simply put, reactive programming is nothing but programming with asynchronous data streams. By applying various operations on stream, we can achieve different computational goals. The primary task in a reactive program is to convert data into streams, regardless of what the source of the data is. While writing modern graphical user interface applications, we process mouse move-and-click events. Currently, most systems get a callback and process these events as and when they happen. Most of the time, the handler does a series of filtering operations before it invokes the action methods associated with the event calls. In this particular context, reactive programming helps us in aggregating the mouse move-and-click events into a collection and sets a filter on them before notifying the handler logic. In this way, the application/handler logic does not get executed unnecessarily.

The stream-processing model is well known, and it is very easy to encode by application developers. Pretty much anything can be converted into a stream. Such candidates include messages, logs, properties, Twitter feeds, blog posts, RSS feeds, and so on. Functional programming techniques are really good at processing streams. A language such as Modern C++, with excellent support for Object/Functional programming, is a natural choice for writing reactive programs. The basic idea behind reactive programming is that there are certain datatypes that represent a value over time. These datatypes (or rather data sequences) are represented as Observable sequences in this programming paradigm. Computations that involve these changing (time-dependent) values will, in turn, themselves have values that change over time, and will need to asynchronously receive notifications (as and when the dependent data changes).

Almost all modern programming languages support functional programming constructs. Functional programming constructs such as Transform, Apply, Filter, Fold, and so on are good for processing streams. Programming asynchronous data streams using functional programming constructs are generally called functional reactive programming (for all practical purposes). The definition given here is an operational one. Consult the work done by Conal Elliott and Paul Hudak as part of the Haskell community to understand the strict definition. Mixing Reactive Programming with FP is gaining traction among developers these days. The Emergence of libraries such as Rx.Net, RxJava, RxJs, and RxCpp and so on is a testimony to this.

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:

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.

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.

Reactive programs can be classified as push-based and pull-based. The pull-based system waits for a demand to push the data streams to the requestor (or subscriber in our case). This is the classic case where the data source is actively polled for more information. This employs the iterator pattern, and IEnumerable <T>/IEnumerator <T> interfaces are specifically designed for such scenarios that are synchronous in nature (the application can block while pulling data). On the other hand, a push-based system aggregates events and pushes through a signal network to achieve the computation. In this case, unlike the pull-based system, data and related updates are handed to the subscriber from the source (Observable sequences in this case). This asynchronous nature is achieved by not blocking the subscriber, but rather making it react to the changes. As you can see, employing this push pattern is more beneficial in rich UI environments where you wouldn't want to block the main UI thread while waiting for some events. This becomes ideal, thus making reactive programs responsive.

The IEnumerable/IObservable duality

If you take a closer look, there is only a subtle difference between these two patterns. IEnumerable<T> can be considered the pull-based equivalent of the push-based IObservable<T>. In fact, they are duals. When two entities exchange information, one entity's pull corresponds to another entity pushing the information. This duality is illustrated in the following diagram:

Let's understand this duality by looking at this sample code, a number sequence generator:

Note

We have striven to use classic C++ constructs to write programs for this particular chapter as there are chapters on Modern C++ language features, language level concurrency, lock-free programming, and related topics for implementing Reactive constructs in Modern C++.

#include <iostream>
#include <vector>
#include <iterator>
#include <memory>
#include "../Common2.h"
using namespace std;

class ConcreteEnumberable : public IEnumerable<int>
{
      int *numberlist,_count;
public:
      ConcreteEnumberable(int numbers[], int count):
            numberlist(numbers),_count(count){}
      ~ConcreteEnumberable() {}

      class Enumerator : public IEnumerator<int>
      {
      int *inumbers, icount, index;
      public:
      Enumerator(int *numbers,
            int count):inumbers(numbers),icount(count),index(0) {}
      bool HasMore() { return index < icount; }
      //---------- ideally speaking, the next function should throw
      //---------- an exception...instead it just returns -1 when the 
      //---------- bound has reached
      int next() { return (index < icount) ?
                   inumbers[index++] : -1; }
      ~Enumerator() {}
      };
      IEnumerator<int> *GetEnumerator()
            { return new Enumerator(numberlist, _count); }
};

The preceding class takes an array of integers as a parameter and we can enumerate over the elements as we have implemented the IEnumerable<T> interface. The Enumeration logic is implemented by the nested class, which implements the IEnumerator<T> interface:

int main()
{
      int x[] = { 1,2,3,4,5 };
      //-------- Has used Raw pointers on purpose here as we have
      //------- not introduced unique_ptr,shared_ptr,weak_ptr yet
      //-------- using auto_ptr will be confusting...otherwise
      //-------- need to use boost library here... ( an overkill)
      ConcreteEnumberable *t = new ConcreteEnumberable(x, 5);
      IEnumerator<int> * numbers = t->GetEnumerator();
      while (numbers->HasMore())
            cout << numbers->next() << endl;
      delete numbers;delete t;
      return 0;
}

The main program instantiates an implementation of the ConcreteEnuerable class and walks through each element.

We will write an even number sequence generator to demonstrate how these data types work together in converting a pull-based program to a push program. The robustness aspect is given low priority to keep the listing terse:

#include "stdafx.h"
#include <iostream>
#include <vector>
#include <iterator>
#include <memory>
#include "../Common2.h"
using namespace std;

class EvenNumberObservable : IObservable<int>{
      int *_numbers,_count;
public:
      EvenNumberObservable(int numbers[],
            int count):_numbers(numbers),_count(count){}
      bool Subscribe(IObserver<int>& observer){
            for (int i = 0; i < _count; ++i)
                  if (_numbers[i] % 2 == 0)
                        observer.OnNext(_numbers[i]);
            observer.OnCompleted();
            return true;
      }
};

The preceding program takes an array of integers, filters out of the odd numbers, and notifies Observer<T> if an even integer is encountered. In this particular case, the data source is pushing data to observer. The implementation of Observer<T> is given as follows:

class SimpleObserver : public IObserver<int>{
public:
      void OnNext(int value) { cout << value << endl; }
      void OnCompleted() { cout << _T("hello completed") << endl; }
      void OnError( CustomException * ex) {}
};

The SimpleObserver class implements the IObserver<T> interface and it has the capability to receive notifications and react to them:

int main()
{
      int x[] = { 1,2,3,4,5 };
      EvenNumberObservable *t = new EvenNumberObservable(x, 5);
      IObserver<int>> *xy = new SimpleObserver();
      t->Subscribe(*xy);
      delete xy; delete t;
      return 0;
}

From the preceding example, you see how one can naturally subscribe for even numbers from an Observable sequence of natural numbers. The system will automatically push (publish) the values to the observer (subscriber) when an even number is detected. The code gives explicit implementations for key interfaces so that one can understand, or speculate what really happens under the hood.

We have now understood how one can convert an IEnumerable<T>-based pull program to an IObservable<T>/IObserver<T>-based push program. In real life, the event source is not as simple as we found in the number stream example given earlier. Let us see how we can convert a MouseMove event into a stream with a small MFC program:

#include "stdafx.h"
#include <afxwin.h>
#include <afxext.h>
#include <math.h>
#include <vector>
#include "../Common2.h"

using namespace std;
class CMouseFrame :public CFrameWnd,IObservable<CPoint>
{
private:
      RECT _rect;
      POINT _curr_pos;
      vector<IObserver<CPoint> *> _event_src;
public:
      CMouseFrame(){
            HBRUSH brush =
                  (HBRUSH)::CreateSolidBrush(RGB(175, 238, 238));
            CString mywindow = AfxRegisterWndClass(
                  CS_HREDRAW | CS_VREDRAW | CS_DBLCLKS,
                  0, brush, 0);
            Create(mywindow, _T("MFC Clock By Praseed Pai"));
      }

The preceding part of the code defines a Frame class that derives from the MFC library the CFrameWnd class and also implements the IObservable<T> interface to force the programmer to implement the Subscribe method. A vector of IObserver<T> will store the list of observers or Subscribers. For this example, we will have only one observer. There is no restriction on the number of observer in the code:

      virtual bool Subscribe(IObserver<CPoint>& observer) {
            _event_src.push_back(&observer);
            return true;
      }

The Subscribe method just stores the reference to the observer onto a vector and returns true: when the mouse is moved, we get notification from the MFC library and if it is in a rectangular area, observer will be notified (the notification code is as follows):

      bool FireEvent(const CPoint& pt) {
            vector<IObserver<CPoint> *>::iterator it =
                  _event_src.begin();
            while (it != _event_src.end()){
                  IObserver<CPoint> *observer = *it;
                  observer->OnNext(pt);
                  //---------- In a Real world Rx programs there is a 
                  //--------- sequence stipulated to call methods...
                  //--------- OnCompleted will be called only when 
                  //--------- all the data is processed...this code
                  //--------- is written to demonstrate the call schema
                  observer->OnCompleted();
                  it++;
            }
            return true;
      }

The FireEvent method walks through the observer's and calls the OnNext method of the observer. It also calls the OnCompleted method of each instance of Observer's: The Rx dispatching mechanism follows certain rules while calling the observer methods. If OnComplete method is called, no more OnNext will be called on the same observer. Similarly, if OnError is called, no further messages will be dispatched to the observer. If we need to follow the conventions stipulated by the Rx model here, the listing will get complicated. The purpose of the code given here is to show how the Rx programming model works in a schematic manner.

      int OnCreate(LPCREATESTRUCT l){
            return CFrameWnd::OnCreate(l);
      }
      void SetCurrentPoint(CPoint pt) {
            this->_curr_pos = pt;
            Invalidate(0);
      }

The SetCurrentPoint method is invoked by observer to set the current point where the text has to be drawn. The Invalidate method is invoked to trigger a WM_PAINT message and the MFC subsystem will route it to OnPaint (as it is wired in the Message maps):

      void OnPaint()
      {
            CPaintDC d(this);
            CBrush b(RGB(100, 149, 237));
            int x1 = -200, y1 = -220, x2 = 210, y2 = 200;
            Transform(&x1, &y1); Transform(&x2, &y2);
            CRect rect(x1, y1, x2, y2);
            d.FillRect(&rect, &b);
            CPen p2(PS_SOLID, 2, RGB(153, 0, 0));
            d.SelectObject(&p2);

            char *str = "Hello Reactive C++";
            CFont f;
            f.CreatePointFont(240, _T("Times New Roman"));
            d.SelectObject(&f);
            d.SetTextColor(RGB(204, 0, 0));
            d.SetBkMode(TRANSPARENT);
            CRgn crgn;
            crgn.CreateRectRgn(rect.left,rect.top,
            rect.right ,rect.bottom);
            d.SelectClipRgn(&crgn);
            d.TextOut(_curr_pos.x, _curr_pos.y,
            CString(str), strlen(str));
      }

The OnPaint method is invoked by the MFC framework when the Invalidate call is made. The method draws the literal string, Hello Reactive C++, on the screen:

      void Transform(int *px, int *py) {
            ::GetClientRect(m_hWnd, &_rect);
            int width = (_rect.right - _rect.left) / 2,
            height = (_rect.bottom - _rect.top) / 2;
           *px = *px + width; *py = height - *py;
      }

The Transform method computes the bound of the client area of the Frame and converts Cartesian coordinates to devise coordinates. This computation can be better done through world coordinate transformations:

      void OnMouseMove(UINT nFlags, CPoint point)
      {
            int x1 = -200,y1= -220, x2 = 210,y2 = 200;
            Transform(&x1, &y1);Transform(&x2, &y2);
            CRect rect(x1, y1, x2, y2);
            POINT pts;
            pts.x = point.x; pts.y = point.y;
            rect.NormalizeRect();
            //--- In a real program, the points will be aggregated
            //---- into a list (stream)
            if (rect.PtInRect(point)) {
                  //--- Ideally speaking this notification has to go
                  //--- through a non blocking call
                  FireEvent(point);
            }
      }

The OnMouseMove method checks whether the mouse position is within a rectangle centered inside the screen and fires the notification to the observer:

      DECLARE_MESSAGE_MAP();
};

BEGIN_MESSAGE_MAP(CMouseFrame, CFrameWnd)
      ON_WM_CREATE()
      ON_WM_PAINT()
      ON_WM_MOUSEMOVE()
END_MESSAGE_MAP()
class WindowHandler : public IObserver<CPoint>
{
private:
      CMouseFrame *window;
public:
      WindowHandler(CMouseFrame *win) : window(win) { }
      virtual ~WindowHandler() { window = 0; }
      virtual void OnCompleted() {}
      virtual void OnError(CustomException *exception) {}
      virtual void OnNext(CPoint value) {
            if (window) window->SetCurrentPoint(value);
      }
};

The preceding class WindowHandler implements the IObserver<T> interface and handles the event notified by CMouseFrame, which implements the IObservable<CPoint> interface. In this canned example, we set the current point by invoking the SetCurrentPoint method to draw the string at the mouse position:

class CMouseApp :public CWinApp
{
      WindowHandler *reactive_handler;
public:
      int InitInstance(){
            CMouseFrame *p = new CMouseFrame();
            p->ShowWindow(1);
            reactive_handler = new WindowHandler(p);
            //--- Wire the observer to the Event Source
            //--- which implements IObservable<T>
            p->Subscribe(*reactive_handler);
            m_pMainWnd = p;
            return 1;
      }
      virtual ~CMouseApp() {
            if (reactive_handler) {
                  delete reactive_handler;
                  reactive_handler = 0;
           }
      }
};

CMouseApp a;

The purpose of this chapter is to introduce readers to the key interfaces of the reactive programming mode they are—IObservable<T> and IObserver<T>. They are in fact the duals of the IEnumerable<T> and IEnumerator<T> interface. We learned how to model those interfaces in classic C++ (well, mostly) and had toy implementations of all of them. Finally, we implemented a GUI program that captures mouse movements and notifies a list of Observers. These toy implementations are to get our feet wet with the ideas and ideals of the Reactive programming model. Our implementations can be considered as implementing of OOP-based reactive programming.

To be proficient in C++ reactive programming, a programmer has to be comfortable with the following topics:

This chapter was mostly about key idioms and why we require a robust model for handling asynchronous data. The next three chapters will cover language features of Modern C++, handling concurrency/parallelism with C++ standard constructs, and lock-free programming (made possible by memory model guarantees). The preceding topics will give the user a firm foundation from which to master functional reactive programming.

In Chapter 5, Introduction to Observables, we will once again return to the topic of Observables and implement interfaces in a functional manner to reiterate some of the concepts. In Chapter 6, Introduction to Event Stream Programming Using C++, we will move towards the advanced event stream processing topics with the help of two industrial-strength libraries that use the Domain Specific Embedded Language (DSEL) approach towards event stream processing.

By now, the stage will be set for the user to be exposed to the industrial-strength RxCpp library and its nuances to write professional-quality Modern C++ programs. In Chapter 7, Introduction to Data Flow Computation and the RxCpp Library and Chapter 8, RxCpp – the Key Elements, we will cover this wonderful library. The following chapters will cover Reactive GUI programming using the Qt library and advanced operators in RxCpp.

The last three chapters cover advanced topics of Reactive design patterns, micro-services in C++, and handling errors/exceptions. By the end of the book, the reader who started with classic C++ will have covered a lot of ground, not only in writing Reactive programs but in the C++ language itself. Because of the nature of the topic, we will cover most of the features of C++ 17 (at the time of writing).

In this chapter, we learned about some key data structures of the Rx programming model. We implemented toy versions of them to familiarize us with the conceptual nuances underpinning them. We started with how GUI events were handled by Windows API, XLib API, MFC, and Qt. We briefly touched upon how events are handled in COM+/CORBA as well. Then, a quick overview of Reactive programming was given. After introducing some interfaces, we implemented them from scratch. Finally, a GUI version of these interfaces on top of MFC was implemented for the sake of completeness. We also dealt with the key philosophical aspects of the book.

In the next chapter, we will make a whirlwind tour of the key features of Modern C++ (C++ Versions 11/14/17) by emphasizing on move semantics, Lambdas, type inference, range-based loops, pipe-able operators, smart pointers, and so on. This is essential for writing even basic code for Reactive Programming.