PHP Reactive Programming

Imperative programming is a programming paradigm around executing statements that change the program's state.

What this means in human language:

Declarative programming is a paradigm focused on describing a program's logic instead of particular executional steps. In other words, in declarative programming, we define what we want instead of how we want it. In contrast to imperative programming, programs in declarative programming are defined with expressions instead of statements.

Very common examples could be SQL and HTML languages. Consider the following database query:

SELECT * FROM user WHERE id = 42 

In SQL, we define what data from what table we want to query, but the implementation details are completely hidden for us. We don't even want to worry about how the database engine stores or indexes the data.

In HTML, we define the structure of elements; what's behind the browser's rendering process isn't important for us. We just want to see the page on the screen.

We can think of sequential and parallel programming as counterparts.

In sequential programming, we're executing processes in order. This means that a process is started when the preceding process has finished. In other words, there is always only one process being executed. The following figure illustrates this principle:

In parallel programming, multiple processes can be executed concurrently:

To make this easier to understand and more relevant to PHP, we can, instead of processes, think of lines of code. PHP interpreter is always sequential and it never executes code in parallel.

In Chapter 9, Multithreaded and Distributed Computing with pthreads and Gearman, we'll use PHP module pthreads that makes it possible to run PHP code in multiple threads, but we'll see that it's not as simple as it seems. Module pthreads, in fact, creates multiple independent PHP interpreters, each running in a separate thread.

The term asynchronous programming is very common in languages such as JavaScript. A very general definition is that, in asynchronous programming, we're executing code in a different order than it was defined. This is typical for any event based application.

For example, in JavaScript, we first define an event listener with its handler, which is executed some time later, when an appropriate event occurs.

In PHP, this could be, for example, a web application that needs to send an e-mail when we create a new blog article. Just, instead of lines of code, we're considering tasks. The following figure demonstrates an asynchronously triggered event:

While the web application was saving an article (processing a task), it triggered an event that sent an e-mail and then carried on with the original task. The event handler had to be defined somewhere before we started this task.

The functional programming paradigm treats program flow as an evaluation of functions. It utilizes several concepts, where the most important for us are eliminating side effects, avoiding mutable data, functions as first-class citizens and higher-order functions. The output of each function is dependent only on its input argument values, therefore, calling the same function twice has to always return the same value. It's based on declarative programming, in the sense of using expressions instead of statements.

Let's have a deeper look what this means:

function sum($array) { 
    $sum = 0; 
    foreach ($array as $value) { 
        $sum += $value; 
    } 
    saveToDatabase($sum); 
    return $sum; 
} 
sum([5, 1, 3, 7, 9]); 

function add($first, $second) { 
    return $first + $second; 
} 
add(5, 2); 

function greaterThan($collection, $threshold) { 
    $out = []; 
    foreach ($collection as $val) { 
        if ($val > $threshold) { 
            $out[] = $val; 
        } 
    } 
    return $out; 
} 
greaterThan([5, 12, 8, 9, 42], 8); 
// will return: [12, 9, 42] 

$input = ['apple', 'banana', 'orange', 'raspberry']; 
$lengths = array_map(function($item) { 
    return strlen($item); 
}, $input); 
// $lengths = [5, 6, 6, 9]; 

array_map(function($item) { 
    return strlen($item); 
}, $input); 

$result = []; 
foreach ($input as $value) { 
    $result[] = strlen($value); 
} 

$input = ['apple', 'banana', 'orange', 'raspberry']; 
$sum = 0; 
foreach ($input as $fruit) { 
    $length = strlen($fruit); 
    if ($length > 5) { 
        $sum += $length; 
    } 
} 
// $sum = 21 
printf("sum: %d\n", $sum); 

$lengths = array_map(function($fruit) { 
    return strlen($fruit); 
}, $input); 
$filtered = array_filter($lengths, function($length) { 
    return $length > 5; 
}); 
$sum = array_reduce($filtered, function($a, $b) { 
    return $a + $b; 
}); 

$sum = array_reduce(array_filter(array_map(function($fruit) { 
    return strlen($fruit); 
}, $input), function($length) { 
    return $length > 5; 
}), function($a, $b) { 
    return $a + $b; 
}); 

array array_map(callable $callback, array $array1 [, $... ]) 
array array_filter(array $array, callable $callback) 
mixed array_reduce(array $array, callable $callback) 

use function Functional\map; 
use function Functional\filter; 
use function Functional\reduce_left; 
 
$sum = reduce_left(filter(map($input, function($fruit) { 
    return strlen($fruit); 
}), function($length) { 
    return $length > 5; 
}), function($val, $i, $col, $reduction) { 
    return $val + $reduction; 
}); 

var sum = inputs 
    .map(fruit => fruit.length) 
    .filter(len => len > 5) 
    .reduce((a, b) => a + b);  

$count = function() { 
    printf("%d ", count($this->fruits)); 
}; 
var_dump(get_class($count)); 
// string(7) "Closure" 

class MyClass { 
    public $fruits; 
    public function __construct($arr) { 
        $this->fruits = $arr; 
    } 
} 

// closures_01.php 
// ... the class definition goes here 
$count = function() { 
    printf("%d ", count($this->fruits)); 
}; 
 
$obj1 = new MyClass(['apple', 'banana', 'orange']); 
$obj2 = new MyClass(['raspberry', 'melon']); 
 
$count->call($obj1); 
$count->call($obj2); 

$ php closures_01.php
3
2

// closures_03.php 
$str = 'Hello, World'; 
 
$func = function() use ($str) { 
    $str .= '!!!'; 
    echo $str . "\n"; 
}; 
$func(); 
echo $str . "\n"; 
 
$func2 = function() use (&$str) { 
    $str .= '???'; 
    echo $str . "\n"; 
}; 
$func2(); 
echo $str . "\n"; 

$ php closures_03.php
Hello, World!!!
Hello, World
Hello, World???
Hello, World???

Reactive programming is yet another programming paradigm. It is based around the ability to easily express data flows and the automatic propagation of changes.

Let's explore this in more depth:

As we can see, the definition is very broad.

The first part about data flows and propagation of change looks like the observer design pattern with iterables. Expressing data flows with ease could be done with functional programming. This all basically describes what we've already seen in this chapter.

The main differences to the observer pattern are how we think and manipulate with data streams. In previous examples, we always worked with arrays as inputs, which are synchronous, while data streams can be both synchronous and asynchronous. From our point of view, it doesn't matter.

Let's see what a typical implementation of the observer pattern might look like in PHP:

// observer_01.php 
class Observable { 
    /** @var Observer[] */ 
    private $observers = []; 
    private $id; 
    static private $total = 0; 
 
    public function __construct() { 
        $this->id = ++self::$total; 
    } 
 
    public function registerObserver(Observer $observer) { 
        $this->observers[] = $observer; 
    } 
 
    public function notifyObservers() { 
        foreach ($this->observers as $observer) { 
            $observer->notify($this, func_get_args()); 
        } 
    } 
 
    public function __toString() { 
        return sprintf('Observable #%d', $this->id); 
    } 
} 

In order to be notified about any changes made by the Observable, we need another class called Observer that subscribes to an Observable:

// observer_01.php 
class Observer { 
    static private $total = 0; 
    private $id; 
 
    public function __construct(Observable $observable) { 
        $this->id = ++self::$total; 
        $observable->registerObserver($this); 
    } 
 
    public function notify($obsr, $args) { 
        $format = "Observer #%d got "%s" from %s\n"; 
        printf($format, $this->id, implode(', ', $args), $obsr); 
    } 
} 

Then, a typical usage might look like the following:

$observer1 = new Observer($subject); 
$observer2 = new Observer($subject); 
$subject->notifyObservers('test'); 

This example will print two messages to the console:

$ php observer_01.php
// Observer #1 got "test" from Observable #1
// Observer #2 got "test" from Observable #1

This almost follows how we defined the reactive programming paradigm. A data stream is a sequence of events coming from an Observable, and changes are propagated to all listening observers. The last point we mentioned above - being able to easily express data flows - isn't really there. What if we wanted to filter out all events that don't match a particular condition, just like we did in the examples with array_filter() and functional programming? This logic would have to go into each Observer class implementation.

The principles of reactive programming are actually very common in some libraries. We'll have a look at three of them and see how these relate to what we've just learned about reactive and functional programming.

$.get('/foo/bar').done(response => { 
    // ... 
}).fail(response => { 
    // ... 
}).complete(response => { 
    // ... 
}); 

// promises_01.js 
function functionReturningAPromise() { 
    var d = $.Deferred(); 
    setTimeout(() => d.resolve(42), 0); 
    return d.promise(); 
} 
 
functionReturningAPromise() 
    .then(value => value + 1) 
    .then(value => 'result: ' + value) 
    .then(value => console.log(value)); 

$ node promises_01.js 
result: 43

gulp.src('src/*.js') 
  .pipe(concat('all.min.js')) 
  .pipe(gulp.dest('build')); 

var filter = require('gulp-filter'); 
var stream = gulp.src('src/*.js'); 
var substream1 = stream.pipe(filter(['*.min.js'])); 
var substream2 = stream.pipe(filter(['!/app/*'])); 

var merge = require('merge2'); 
merge(gulp.src('src/*.js'), gulp.src('vendor/*')) 
    .pipe(uglify()); 
    .pipe(gulp.dest('build')); 

use Symfony\Component\EventDispatcher\EventDispatcher; 
$dispatcher = new EventDispatcher(); 
$listener = new AcmeListener(); 
$dispatcher->addListener('event_name', [$listener, 'action']); 

Now that we've seen that the principles in the reactive programming paradigm aren't completely new for us, we can start thinking about how to put all this together. In other words, what libraries or frameworks do we really need in order to start writing reactive code.

Reactive Extensions (ReactiveX or just Rx in short) are a set of libraries in various languages that make reactive programming easy even in languages where concepts of asynchronous and functional programming are clumsy, such as PHP. However, there's a very important distinction:

Reactive programming doesn't equal Reactive Extensions.

A Reactive Extension is a library that introduces certain principles as one of the possible ways to approach reactive programming. Very often, when somebody tells you they're using reactive programming to do something in their applications, they're in fact talking about a particular Reactive Extension library in their favorite language.

Reactive Extensions were originally made by Microsoft for .NET and called Rx.NET. Later, it was ported by Netflix to Java as RxJava. Now, there are over a dozen supported languages, the most popular probably being RxJS - the JavaScript implementation.

All ports follow a very similar API design, however, differences occur and we'll talk about them a couple of times. We'll be mostly interested in differences between RxPHP and RxJS.

RxPHP is mostly uncharted territory. A more typical environment where we encounter asynchronous events is JavaScript, so we'll first demonstrate examples in JavaScript (and RxJS 5), and afterwards we will have a look at RxPHP.

function searchAndReturnPromise(term) { 
    // perform an AJAX request and return a Promise 
} 
 
var keyup = Rx.Observable.fromEvent($('#textInput'), 'keyup') 
    .map(e => e.target.value) 
    .filter(text => text.length > 2) 
    .debounceTime(750) 
    .distinctUntilChanged(); 
var searcher = keyup.switchMap(searchAndReturnPromise); 

var mouseup   = Rx.Observable.fromEvent(dragTarget, 'mouseup'); 
var mousemove = Rx.Observable.fromEvent(document, 'mousemove'); 
var mousedown = Rx.Observable.fromEvent(dragTarget, 'mousedown'); 
 
var mousedrag = mousedown.mergeMap(md => { 
    var sX = md.offsetX, sY = md.offsetY; 
    return mousemove.map(mm => { 
        mm.preventDefault(); 
        return {left: mm.clientX - sX, top: mm.clientY - sY}; 
    }).takeUntil(mouseup); 
}); 
 
var subscription = mousedrag.subscribe(pos => { 
    dragTarget.style.top = pos.top + 'px'; 
    dragTarget.style.left = pos.left + 'px'; 
}); 

RxPHP ( https://github.com/ReactiveX/RxPHP ) is a port of RxJS. We're going to be using Composer to handle all dependencies in our PHP projects. It has become a state of the art tool, so if you haven't used it before, download it first and check out some basic usage at  https://getcomposer.org/ .

Then, create a new directory and initialize a composer project:

$ mkdir rxphp_01
$ cd rxphp_01
$ php composer.phar init

Fill in the required fields by the interactive wizard and then add RxPHP as a dependency:

$ php composer.phar require reactivex/rxphp

When the library successfully downloads, composer will also create autoload.php file to handle all class auto-loading on demand.

Then, our code will print string lengths of different types of fruit:

// rxphp_01.php 
require __DIR__ . '/vendor/autoload.php'; 
 
$fruits = ['apple', 'banana', 'orange', 'raspberry']; 
$observer = new \Rx\Observer\CallbackObserver( 
    function($value) { 
        printf("%s\n", $value); 
    }, null, function() { 
        print("Complete\n"); 
    }); 
 
\Rx\Observable::fromArray($fruits) 
    ->map(function($value) { 
        return strlen($value); 
    }) 
    ->subscribe($observer); 

We first created an observer - CallbackObserver to be precise - which takes three functions as arguments. These are called on the next item in the stream, on error and when the input stream is complete and won't emit any more items.

The advantage of the CallbackObserver class is that we don't need to write a custom observer class every time we want to handle incoming items in some special and not very reusable way. With CallbackObserver, we can just write the callables for signals we want to handle.

When we run this example, we'll see:

$ php rxphp_01.php 
5
6
6
9
Complete

This example was very easy, but compared to the JavaScript environment, it's not very common to use asynchronous operations in PHP and, in case we do have to work asynchronously, it's probably something non-trivial. In Chapter 3, Writing a Reddit reader with RxPHP, we'll use Symfony Console component to handle all user input from the command line and, where we can, use similar principles to handling mouse events as we saw in the two RxJS examples above.

The JavaScript examples work very well as examples of what reactive programming using Reactive Extensions looks like and what its benefits are.

In this chapter, we tried to explain the common programming paradigms used in most programming languages. These were: imperative, declarative and functional programming. We also compared the meanings of asynchronous and parallel code.

We spent some time on practical examples of functional programming in PHP and its downsides, and we went through examples of some not very common features, such as the Closure class.

Then, we examined the definition of reactive programming and how it's related to all we saw previously in this chapter.

We introduced Reactive Extensions (Rx) as a library for one of the possible approaches to reactive programming.

In two examples of RxJS, we saw what working with Reactive Extensions looks like in practice and how this matches our definition of reactive programming.

Finally, we introduced RxPHP, which we'll use throughout this entire book. We also quickly talked about differences between RxPHP 1.x and RxPHP 2.

In the next chapter, we'll have a closer look at various parts of the RxPHP library and talk more about the principles used in Reactive Extensions.