src/Observable.php
public function do($onNextOrObserver = null, callable $onError = null, callable $onCompleted = null): Observable
    {
        if ($onNextOrObserver instanceof ObserverInterface) {
            $observer = $onNextOrObserver;
        } elseif (is_callable($onNextOrObserver)) {
            $observer = new DoObserver($onNextOrObserver, $onError, $onCompleted);
        } else {
            throw new \InvalidArgumentException('The first argument needs to be a "callable" or "Observer"');
        }

        return $this->lift(function () use ($observer) {
            return new DoOnEachOperator($observer);
        });
    }
public function do($onNextOrObserver = null, callable $onError = null, callable $onCompleted = null): Observable
    {
        if ($onNextOrObserver instanceof ObserverInterface) {
            $observer = $onNextOrObserver;
        } elseif (is_callable($onNextOrObserver)) {
            $observer = new DoObserver($onNextOrObserver, $onError, $onCompleted);
        } else {
            throw new \InvalidArgumentException('The first argument needs to be a "callable" or "Observer"');
        }

        return $this->lift(function () use ($observer) {
            return new DoOnEachOperator($observer);
        });
    }
public function do ( $ onNextOrObserver = null , callable $ onError = null , callable $ onCompleted = null ) : Observable { if ( $ onNextOrObserver instanceof ObserverInterface ) { $ observer = $ onNextOrObserver ; } elseif ( is_callable ( $ onNextOrObserver ) ) { $ observer = new DoObserver ( $ onNextOrObserver , $ onError , $ onCompleted ) ; } else { throw new \ InvalidArgumentException ( 'The first argument needs to be a "callable" or "Observer"' ) ; } return $ this -> lift ( function ( ) use ( $ observer ) { return new DoOnEachOperator ( $ observer ) ; } ) ; }
Invokes an action for each element in the observable sequence and invokes an action upon graceful
or exceptional termination of the observable sequence.
This method can be used for debugging, logging, etc. of query behavior by intercepting the message stream to
run arbitrary actions for messages on the pipeline.

When using do, it is important to note that the Observer may receive additional
events after a stream has completed or errored (such as when using a repeat or resubscribing).
If you are using an Observable that extends the AbstractObservable, you will not receive these
events. For this special case, use the DoObserver.

doOnNext, doOnError, and doOnCompleted uses the DoObserver internally and will receive these
additional events.

@param callable|ObserverInterface $onNextOrObserver
@param callable $onError
@param callable $onCompleted
@return Observable
@throws \InvalidArgumentException

@demo do/do.php
@operator
@reactivex do
Invokes an action for each element in the observable sequence and invokes an action upon graceful or exceptional termination of the observable sequence . This method can be used for debugging logging etc . of query behavior by intercepting the message stream to run arbitrary actions for messages on the pipeline .