Let's have a look at a very simple example of RxPHP, similar to what we did in the previous chapter, and use it to demonstrate some of the basic principles behind Reactive Extensions.

We won't bother with defining an observer right now and will focus only on Observables and operators:

// rxphp_basics_01.php 
use Rx\Observable; 
$fruits = ['apple', 'banana', 'orange', 'raspberry']; 
 
Observable::fromArray($fruits) // Observable 
    ->map(function($value) { // operator 
        return strlen($value); 
    }) 
    ->filter(function($len) { // operator 
        return $len > 5; 
    }) 
    ->subscribe($observer); // observer 

In this example, we have one Observable, two operators and one observer.

An Observable can be chained with operators. In this example, the operators are map() and filter().

Observables have the subscribe() method that is used by observers to start receiving values...

When talking about Observables, we use terms such as emit/send value/item. Commonly, we say that an Observable emits an item, but we understand the same from an Observable sends a value as well.

By emit/send we mean that an Observable is calling the onNext method on an observer.

When talking about Observables, we use terms such as send error/complete notification/signal. We also often mention that an Observable completes, which means that an Observable has sent a complete notification.

By notification/signal we mean that an Observable is calling the onError or onComplete method on an observer.

In the preceding paragraph, we worked with a simple RxPHP demo that had one Observable, two operators and one observer.

This structure formed an operator/Observable chain. We'll understand the same thing from both of the terms operator chain and Observable chain (sometimes also referred to as a chain of Observable operators). This is because, from our perspective...

Since this chapter is going to be mostly about Observables, observers and operators, we're going to start with them.

We've already seen a sneak peak in this chapter, and now we'll go into more detail.

One common use case for Subject class is proxying all values and notifications from its source Observable.

In one of the preceding paragraphs, we wrote the PrintObserver class, which prints all values it receives. However, a more common situation is where we want to output values from an Observable while being able to chain it with another operator or observer. The Subject class exactly fits this use case, so we'll rewrite the preceding PrintObserver class and inherit Subject instead of AbstractObserver:

class DebugSubject extends Rx\Subject\Subject { 
  public function __construct($identifier=null, $maxLen=64){ 
    $this->identifier = $identifier; 
    $this->maxLen = $maxLen; 
  } 
  public function onCompleted() { 
    printf("%s%s onCompleted\n", $this->getTime(), $this->id());
    parent::onCompleted(); 
  }  
  public function onNext($val) { 
    $type = is_object($val) ? get_class($val) : gettype...

We're going to work with calls to remote API's a few times throughout this book, so it would be very handy to have an operator that transforms JSON string responses into their PHP array representations.

This example looks like something that could be easily done with just the map() operator:

// rxphp_06.php  
Rx\Observable::just('{"value":42}') 
    ->map(function($value) { 
        return json_decode($value, true); 
    }) 
    ->subscribe(new DebugSubject()); 

This prints the correct result for sure, as we can see in the following output:

$ php rxphp_06.php
16:39:50 [] onNext: {"value": 42} (array)
16:39:50 [] onCompleted

Well, but what about malformed JSON strings? What happens if we try to decode the following:

Rx\Observable::just('NA') 
    ->map(function($value) { 
        return json_decode($value, true); 
    }) 
    ->subscribe(new DebugSubject()); 

The function json_decode() doesn't throw...