Writing the DebugSubject class

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($val); 
 
    if (is_object($val) && method_exists($val, '__toString')) { 
      $str = (string)$val; 
    } elseif (is_object($val)) { 
      $str = get_class($val); 
    } elseif (is_array($val)) { 
      $str = json_encode($val); 
    } else { 
      $str = $val; 
    } 
 
    if (is_string($str) && strlen($str) > $this->maxLen) { 
      $str = substr($str, 0, $this->maxLen) . '...'; 
    } 
    printf("%s%s onNext: %s (%s)\n", 
        $this->getTime(), $this->id(), $str, $type); 
    parent::onNext($value); 
  } 
  public function onError(Exception $error) { 
    $msg = $error->getMessage(); 
    printf("%s%s onError (%s): %s\n", $this->getTime(),$this-> 
        $this->id(), get_class($error), $msg); 
    parent::onError($error); 
  } 
  private function getTime() { 
    return date('H:i:s'); 
  } 
  private function id() { 
    return ' [' . $this->identifier . ']'; 
  } 
} 

This DebugSubject class prints all values, their types, and the time they were received by the DebugSubject. It also allows us to set a unique identifier for each DebugSubject instance to be able to distinguish their output. We're going to use this class a couple of times throughout this book to quickly see what's going on inside our Observable chains.

Then, using this class is just like using any other observer:

// rxphp_04.php 
$fruits = ['apple', 'banana', 'orange', 'raspberry']; 
$observer = Rx\Observable::fromArray($fruits) 
    ->subscribe(new DebugSubject()); 

The output in the console is as follows:

$ php rxphp_04.php
17:15:21 [] onNext: apple (string)
17:15:21 [] onNext: banana (string)
17:15:21 [] onNext: orange (string)
17:15:21 [] onNext: raspberry (string)
17:15:21 [] onCompleted

Chaining Subjects and operators works just as with Observables:

// rxphp_05.php 
$subject = new DebugSubject(1); 
$subject 
    ->map(function($item) { 
        return strlen($item); 
    }) 
    ->subscribe(new DebugSubject(2)); 
 
$observable = Rx\Observable::fromArray($fruits); 
$observable->subscribe($subject); 

In this example, we first created an instance of DebugSubject, then we chained it with the map() operator, which returns the lengths of each item. Finally, we subscribed another DebugSubject that will print only numbers because it's placed after map(). Then we created an Observable from an array (we've seen this static method previously), which is going to be the source emitting all items. The result is as follows:

17:33:36 [1] onNext: apple (string)
17:33:36 [2] onNext: 5 (integer)
17:33:36 [1] onNext: banana (string)
17:33:36 [2] onNext: 6 (integer)
17:33:36 [1] onNext: orange (string)
17:33:36 [2] onNext: 6 (integer)
17:33:36 [1] onNext: raspberry (string)
17:33:36 [2] onNext: 9 (integer)
17:33:36 [1] onCompleted
17:33:36 [2] onCompleted

Note that the order of messages matches our assumption that the source Observable emits one value at a time, which is propagated through the entire chain.

Note

There's one important side effect of using Subjects as we did that isn't very obvious. Since we subscribe it to the preceding Observable, it turns it from "cold" into "hot", which might be unwanted in some use cases.

RxPHP provides a series of operators all starting with the "doOn" prefix that are intended to be placed inside the operator chain to execute side effects without subscribing to an Observable. We'll have a better look at them in Chapter 5, Testing RxPHP Code.