<commit_before><?php

namespace Rx;

use Rx\Observable\AnonymousObservable;
use Rx\Observable\ArrayObservable;
use Rx\Observable\ConnectableObservable;
use Rx\Observable\EmptyObservable;
use Rx\Observable\ErrorObservable;
use Rx\Observable\IntervalObservable;
use Rx\Observable\IteratorObservable;
use Rx\Observable\MulticastObservable;
use Rx\Observable\NeverObservable;
use Rx\Observable\RangeObservable;
use Rx\Observable\ReturnObservable;
use Rx\Observable\TimerObservable;
use Rx\Observer\CallbackObserver;
use Rx\Observer\DoObserver;
use Rx\Operator\AsObservableOperator;
use Rx\Operator\BufferWithCountOperator;
use Rx\Operator\CatchErrorOperator;
use Rx\Operator\CombineLatestOperator;
use Rx\Operator\ConcatAllOperator;
use Rx\Operator\ConcatMapOperator;
use Rx\Operator\ConcatOperator;
use Rx\Operator\CountOperator;
use Rx\Operator\DefaultIfEmptyOperator;
use Rx\Operator\DeferOperator;
use Rx\Operator\DelayOperator;
use Rx\Operator\DematerializeOperator;
use Rx\Operator\DistinctOperator;
use Rx\Operator\DistinctUntilChangedOperator;
use Rx\Operator\DoOnEachOperator;
use Rx\Operator\GroupByUntilOperator;
use Rx\Operator\MapOperator;
use Rx\Operator\FilterOperator;
use Rx\Operator\MinOperator;
use Rx\Operator\MaxOperator;
use Rx\Operator\MaterializeOperator;
use Rx\Operator\MergeAllOperator;
use Rx\Operator\RaceOperator;
use Rx\Operator\ReduceOperator;
use Rx\Operator\RepeatOperator;
use Rx\Operator\RepeatWhenOperator;
use Rx\Operator\RetryOperator;
use Rx\Operator\RetryWhenOperator;
use Rx\Operator\ScanOperator;
use Rx\Operator\SkipLastOperator;
use Rx\Operator\SkipOperator;
use Rx\Operator\SkipUntilOperator;
use Rx\Operator\StartWithArrayOperator;
use Rx\Operator\SkipWhileOperator;
use Rx\Operator\SubscribeOnOperator;
use Rx\Operator\SwitchFirstOperator;
use Rx\Operator\SwitchLatestOperator;
use Rx\Operator\TakeLastOperator;
use Rx\Operator\TakeOperator;
use Rx\Operator\TakeUntilOperator;
use Rx\Operator\TakeWhileOperator;
use Rx\Operator\ThrottleOperator;
use Rx\Operator\TimeoutOperator;
use Rx\Operator\TimestampOperator;
use Rx\Operator\ToArrayOperator;
use Rx\Operator\ZipOperator;
use Rx\Scheduler\ImmediateScheduler;
use Rx\Subject\AsyncSubject;
use Rx\Subject\BehaviorSubject;
use Rx\Subject\ReplaySubject;
use Rx\Subject\Subject;
use Rx\Disposable\EmptyDisposable;
use Rx\Disposable\CallbackDisposable;

class Observable implements ObservableInterface
{
    protected $observers = [];
    protected $started = false;

    public function subscribe(ObserverInterface $observer, $scheduler = null)
    {
        $this->observers[] = $observer;
        $this->started     = true;

        return new CallbackDisposable(function () use ($observer) {
            $this->removeObserver($observer);
        });
    }

    /**
     * @internal
     */
    public function removeObserver(ObserverInterface $observer)
    {
        $key = array_search($observer, $this->observers, true);

        if (false === $key) {
            return false;
        }

        unset($this->observers[$key]);

        return true;
    }

    public function subscribeCallback(callable $onNext = null, callable  $onError = null, callable $onCompleted = null, SchedulerInterface $scheduler = null)
    {
        $observer = new CallbackObserver($onNext, $onError, $onCompleted);

        return $this->subscribe($observer, $scheduler);
    }

    /**
     * Creates an observable sequence from a specified subscribeAction callable implementation.
     *
     * @param callable $subscribeAction Implementation of the resulting observable sequence's subscribe method.
     * @return AnonymousObservable The observable sequence with the specified implementation for the subscribe method.
     *
     * @demo create/create.php
     * @operator
     * @reactivex create
     */
    public static function create(callable $subscribeAction)
    {
        return new AnonymousObservable($subscribeAction);
    }

    /**
     * Returns an Observable that emits an infinite sequence of ascending integers starting at 0, with a constant interval of time of your choosing between emissions.
     *
     * @param $interval int Period for producing the values in the resulting sequence (specified as an integer denoting milliseconds).
     * @param SchedulerInterface|null $scheduler
     * @return IntervalObservable An observable sequence that produces a value after each period.
     *
     * @demo interval/interval.php
     * @operator
     * @reactivex interval
     */
    public static function interval($interval, $scheduler = null)
    {
        return new IntervalObservable($interval, $scheduler);
    }

    /**
     * Returns an observable sequence that contains a single element.
     *
     * @param mixed $value Single element in the resulting observable sequence.
     * @return \Rx\Observable\AnonymousObservable An observable sequence with the single element.
     *
     * @demo just/just.php
     * @operator
     * @reactivex just
     */
    public static function just($value)
    {
        return new ReturnObservable($value);
    }

    /**
     * Returns an empty observable sequence.
     *
     * @return EmptyObservable An observable sequence with no elements.
     *
     * @demo empty-observable/empty-observable.php
     * @operator
     * @reactivex empty-never-throw
     */
    public static function emptyObservable()
    {
        return new EmptyObservable();
    }

    /**
     * Returns a non-terminating observable sequence, which can be used to denote an infinite duration.
     *
     * @return NeverObservable An observable sequence whose observers will never get called.
     *
     * @demo never/never.php
     * @operator
     * @reactivex empty-never-throw
     */
    public static function never()
    {
        return new NeverObservable();
    }

    /**
     * Returns an observable sequence that terminates with an exception.
     *
     * @param $error
     * @return ErrorObservable The observable sequence that terminates exceptionally with the specified exception object.
     *
     * @demo error-observable/error-observable.php
     * @operator
     * @reactivex empty-never-throw
     */
    public static function error(\Exception $error)
    {
        return new ErrorObservable($error);
    }

    /**
     * Combine an Observable together with another Observable by merging their emissions into a single Observable.
     *
     * @param ObservableInterface $otherObservable
     * @return AnonymousObservable
     *
     * @demo merge/merge.php
     * @operator
     * @reactivex merge
     */
    public function merge(ObservableInterface $otherObservable)
    {
        return self::mergeAll(new AnonymousObservable(function (ObserverInterface $observer, SchedulerInterface $schedule) use ($otherObservable) {
            $observer->onNext($this);
            $observer->onNext($otherObservable);
            $observer->onCompleted();
        }));
    }

    /**
     * Merges an observable sequence of observables into an observable sequence.
     *
     * @param ObservableInterface $sources
     * @return AnonymousObservable
     *
     * @demo merge/merge-all.php
     * @operator
     * @reactivex merge
     */
    public static function mergeAll(ObservableInterface $sources)
    {
        return (new EmptyObservable())->lift(function () use ($sources) {
            return new MergeAllOperator($sources);
        });
    }

    /**
     * Converts an array to an observable sequence
     *
     * @param array $array
     * @return ArrayObservable
     *
     * @demo fromArray/fromArray.php
     * @operator
     * @reactivex from
     */
    public static function fromArray(array $array)
    {
        return new ArrayObservable($array);
    }

    /**
     * Converts an Iterator into an observable sequence
     *
     * @param \Iterator $iterator
     * @return IteratorObservable
     *
     * @demo iterator/iterator.php
     * @operator
     * @reactivex from
     */
    public static function fromIterator(\Iterator $iterator)
    {
        return new IteratorObservable($iterator);
    }

    /**
     * Returns an observable sequence that invokes the specified factory function whenever a new observer subscribes.
     *
     * @param callable $factory
     * @return \Rx\Observable\AnonymousObservable
     *
     * @demo defer/defer.php
     * @operator
     * @reactivex defer
     */
    public static function defer(callable $factory)
    {
        return (new EmptyObservable())->lift(function () use ($factory) {
            return new DeferOperator($factory);
        });
    }

    /**
     * Generates an observable sequence of integral numbers within a specified range, using the specified scheduler to
     * send out observer messages.
     *
     * @param $start
     * @param $count
     * @return RangeObservable
     *
     * @demo range/range.php
     * @operator
     * @reactivex range
     */
    public static function range($start, $count, SchedulerInterface $scheduler = null)
    {
        return new RangeObservable($start, $count, $scheduler);
    }

    /**
     * Invokes the specified function asynchronously on the specified scheduler, surfacing the result through an
     * observable sequence.
     *
     * @param callable $action
     * @param SchedulerInterface $scheduler
     * @return AnonymousObservable
     *
     * @demo start/start.php
     * @operator
     * @reactivex start
     */
    public static function start(callable $action, SchedulerInterface $scheduler = null)
    {
        $scheduler = $scheduler ?: new ImmediateScheduler();
        $subject   = new AsyncSubject();

        $scheduler->schedule(function () use ($subject, $action) {
            $result = null;
            try {
                $result = call_user_func($action);
            } catch (\Exception $e) {
                $subject->onError($e);
                return;
            }
            $subject->onNext($result);
            $subject->onCompleted();
        });

        return $subject->asObservable();
    }

    /**
     * Takes a transforming function that operates on each element.
     *
     * @param callable $selector
     * @return AnonymousObservable
     *
     * @demo map/map.php
     * @operator
     * @reactivex map
     */
    public function map(callable $selector)
    {
        return $this->lift(function () use ($selector) {
            return new MapOperator($selector);
        });
    }

    /**
     * Maps operator variant that calls the map selector with the index and value
     *
     * @param callable $selector
     * @return AnonymousObservable
     *
     * @demo map/mapWithIndex.php
     * @operator
     * @reactivex map
     */
    public function mapWithIndex(callable $selector)
    {
        $index = 0;
        return $this->map(function ($value) use ($selector, &$index) {
            return call_user_func_array($selector, [$index++, $value]);
        });
    }

    /**
     * Maps every value to the same value every time
     *
     * @param $value
     * @return AnonymousObservable
     *
     * @demo map/mapTo.php
     * @operator
     * @reactivex map
     */
    public function mapTo($value)
    {
        return $this->map(function () use ($value) {
            return $value;
        });
    }

    /**
     * Alias for Map
     *
     * @param callable $selector
     * @return \Rx\Observable\AnonymousObservable
     *
     * @operator
     * @reactivex map
     */
    public function select(callable $selector)
    {
        return $this->map($selector);
    }

    /**
     * Emit only those items from an Observable that pass a predicate test.
     *
     * @param callable $predicate
     * @return \Rx\Observable\AnonymousObservable
     *
     * @demo filter/filter.php
     * @operator
     * @reactivex filter
     */
    public function filter(callable $predicate)
    {
        return $this->lift(function () use ($predicate) {
            return new FilterOperator($predicate);
        });
    }

    /**
     * Alias for filter
     *
     * @param callable $predicate
     * @return \Rx\Observable\AnonymousObservable
     *
     * @operator
     * @reactivex filter
     */
    public function where(callable $predicate)
    {
        return $this->filter($predicate);
    }

    /**
     * Projects each element of an observable sequence to an observable sequence and merges the resulting observable sequences into one observable sequence.
     *
     * @param callable $selector
     * @return AnonymousObservable
     *
     * @demo flatMap/flatMap.php
     * @operator
     * @reactivex flatMap
     */
    public function flatMap(callable $selector)
    {
        return self::mergeAll($this->select($selector));
    }

    /**
     * Projects each element of the source observable sequence to the other observable sequence and merges the
     * resulting observable sequences into one observable sequence.
     *
     * @param ObservableInterface $observable - An an observable sequence to project each element from the source
     * sequence onto.
     *
     * @return AnonymousObservable
     *
     * @demo concat/flatMapTo.php
     * @operator
     * @reactivex flatMap
     */
    public function flatMapTo(ObservableInterface $observable)
    {
        return $this->flatMap(function () use ($observable) {
            return $observable;
        });
    }

    /**
     * Alias for flatMap
     *
     * @param $selector
     * @return AnonymousObservable
     *
     * @operator
     * @reactivex flatMap
     */
    public function selectMany($selector)
    {
        return $this->flatMap($selector);
    }

    /**
     * Bypasses a specified number of elements in an observable sequence and then returns the remaining elements.
     *
     * Transform the items emitted by an Observable into Observables, and mirror those items emitted by the
     * most-recently transformed Observable.
     *
     * The flatMapLatest operator is similar to the flatMap and concatMap methods described above, however, rather than
     * emitting all of the items emitted by all of the Observables that the operator generates by transforming items
     * from the source Observable, flatMapLatest instead emits items from each such transformed Observable only until
     * the next such Observable is emitted, then it ignores the previous one and begins emitting items emitted by the
     * new one.
     *
     * @param callable $selector - A transform function to apply to each source element.
     * @return AnonymousObservable - An observable sequence which transforms the items emitted by an Observable into
     * Observables, and mirror those items emitted by the most-recently transformed Observable.
     *
     * @demo flatMap/flatMapLatest.php
     * @operator
     * @reactivex flatMap
     */
    public function flatMapLatest(callable $selector, SchedulerInterface $scheduler = null)
    {
        return $this->map($selector)->switchLatest($scheduler);
    }

    /**
     * @param integer $count
     * @return AnonymousObservable
     *
     * @demo skip/skip.php
     * @operator
     * @reactivex skip
     */
    public function skip($count)
    {
        return $this->lift(function () use ($count) {
            return new SkipOperator($count);
        });
    }

    /**
     * Bypasses elements in an observable sequence as long as a specified condition is true and then returns the
     * remaining elements.
     *
     * @param callable $predicate A function to test each element for a condition.
     *
     * @return AnonymousObservable An observable sequence that contains the elements from the input sequence starting
     * at the first element in the linear series that does not pass the test specified by predicate.
     *
     * @demo skip/skipWhile.php
     * @operator
     * @reactivex skipWhile
     */
    public function skipWhile(callable $predicate)
    {
        return $this->lift(function () use ($predicate) {
            return new SkipWhileOperator($predicate);
        });
    }

    /**
     * Bypasses elements in an observable sequence as long as a specified condition is true and then returns the
     * remaining elements. The element's index is used in the logic of the predicate function.
     *
     * @param callable $predicate A function to test each element for a condition; the first parameter of the
     * function represents the index of the source element, the second parameter is the value.
     *
     * @return AnonymousObservable An observable sequence that contains the elements from the input sequence starting
     * at the first element in the linear series that does not pass the test specified by predicate.
     *
     * @demo skip/skipWhileWithIndex.php
     * @operator
     * @reactivex skipWhile
     */
    public function skipWhileWithIndex(callable $predicate)
    {
        $index = 0;
        return $this->skipWhile(function ($value) use ($predicate, &$index) {
            return call_user_func_array($predicate, [$index++, $value]);
        });
    }

    /**
     * Returns a specified number of contiguous elements from the start of an observable sequence
     *
     * @param integer $count
     * @return AnonymousObservable|EmptyObservable
     *
     * @demo take/take.php
     * @operator
     * @reactivex take
     */
    public function take($count)
    {
        if ($count === 0) {
            return new EmptyObservable();
        }

        return $this->lift(function () use ($count) {
            return new TakeOperator($count);
        });
    }

    /**
     * Returns the values from the source observable sequence until the other observable sequence produces a value.
     *
     * @param ObservableInterface $other - other Observable sequence that terminates propagation of elements of
     * the source sequence.
     * @return AnonymousObservable - An observable sequence containing the elements of the source sequence up to the
     * point the other sequence interrupted further propagation.
     *
     * @demo take/take.php
     * @operator
     * @reactivex take
     */
    public function takeUntil(ObservableInterface $other)
    {
        return $this->lift(function () use ($other) {
            return new TakeUntilOperator($other);
        });
    }

    /**
     * Returns elements from an observable sequence as long as a specified condition is true.  It takes as a parameter a
     * a callback to test each source element for a condition.  The callback predicate is called with the value of the
     * element.
     *
     * @param callable $predicate
     * @return AnonymousObservable
     *
     * @demo take/takeWhile.php
     * @operator
     * @reactivex takeWhile
     */
    public function takeWhile(callable $predicate)
    {
        return $this->lift(function () use ($predicate) {
            return new TakeWhileOperator($predicate);
        });
    }

    /**
     * Returns elements from an observable sequence as long as a specified condition is true.  It takes as a parameter a
     * a callback to test each source element for a condition.  The callback predicate is called with the index and the
     * value of the element.
     *
     * @param callable $predicate
     * @return AnonymousObservable
     *
     * @demo take/takeWhileWithIndex.php
     * @operator
     * @reactivex takeWhile
     */
    public function takeWhileWithIndex(callable $predicate)
    {
        $index = 0;
        return $this->takeWhile(function ($value) use ($predicate, &$index) {
            return call_user_func_array($predicate, [$index++, $value]);
        });
    }

    /**
     * Returns a specified number of contiguous elements from the end of an observable sequence.
     *
     * @param $count
     * @return AnonymousObservable
     *
     * @demo take/takeLast.php
     * @operator
     * @reactivex takeLast
     */
    public function takeLast($count)
    {
        return $this->lift(function () use ($count) {
            return new TakeLastOperator($count);
        });
    }

    /**
     * Groups the elements of an observable sequence according to a specified key selector function and comparer and selects the resulting elements by using a specified function.
     *
     * @param callable $keySelector
     * @param callable|null $elementSelector
     * @param callable|null $keySerializer
     * @return AnonymousObservable
     *
     * @demo groupBy/groupBy.php
     * @operator
     * @reactivex groupBy
     */
    public function groupBy(callable $keySelector, callable $elementSelector = null, callable $keySerializer = null)
    {
        return $this->groupByUntil($keySelector, $elementSelector, function () {

            // observable that never calls
            return new AnonymousObservable(function () {
                // todo?
                return new EmptyDisposable();
            });
        }, $keySerializer);
    }

    /**
     * Groups the elements of an observable sequence according to a specified key selector function and comparer and selects the resulting elements by using a specified function.
     *
     * @param callable $keySelector
     * @param callable|null $elementSelector
     * @param callable|null $durationSelector
     * @param callable|null $keySerializer
     * @return AnonymousObservable
     *
     * @demo groupBy/groupByUntil.php
     * @operator
     * @reactivex groupBy
     */
    public function groupByUntil(callable $keySelector, callable $elementSelector = null, callable $durationSelector = null, callable $keySerializer = null)
    {
        return $this->lift(function () use ($keySelector, $elementSelector, $durationSelector, $keySerializer) {
            return new GroupByUntilOperator($keySelector, $elementSelector, $durationSelector, $keySerializer);
        });
    }

    /**
     * Lifts a function to the current Observable and returns a new Observable that when subscribed to will pass
     * the values of the current Observable through the Operator function.
     *
     * @param callable $operatorFactory
     * @return AnonymousObservable
     */
    public function lift(callable $operatorFactory)
    {
        return new AnonymousObservable(function (ObserverInterface $observer, SchedulerInterface $schedule) use ($operatorFactory) {
            $operator = $operatorFactory();
            return $operator($this, $observer, $schedule);
        });
    }

    /**
     * This method allows the use of extra operators with the namespace:
     * Rx\Operator
     * and also custom operators by adding an operator class with the
     * namespace format:
     * CustomNamespace\Rx\Operator\OperatorNameOperator
     *
     * @param $name
     * @param $arguments
     * @return Observable
     * 
     * @demo custom-operator/rot13.php
     */
    public function __call($name, $arguments)
    {
        $fullNamespace = 'Rx\\Operator\\';
        if ($name[0] === '_') {
            list($_, $namespace, $methodName) = explode('_', $name);
            $name = $methodName;<commit_msg>Added array type hint to Observable::__call (#109)<commit_after><?php

namespace Rx;

use Rx\Observable\AnonymousObservable;
use Rx\Observable\ArrayObservable;
use Rx\Observable\ConnectableObservable;
use Rx\Observable\EmptyObservable;
use Rx\Observable\ErrorObservable;
use Rx\Observable\IntervalObservable;
use Rx\Observable\IteratorObservable;
use Rx\Observable\MulticastObservable;
use Rx\Observable\NeverObservable;
use Rx\Observable\RangeObservable;
use Rx\Observable\ReturnObservable;
use Rx\Observable\TimerObservable;
use Rx\Observer\CallbackObserver;
use Rx\Observer\DoObserver;
use Rx\Operator\AsObservableOperator;
use Rx\Operator\BufferWithCountOperator;
use Rx\Operator\CatchErrorOperator;
use Rx\Operator\CombineLatestOperator;
use Rx\Operator\ConcatAllOperator;
use Rx\Operator\ConcatMapOperator;
use Rx\Operator\ConcatOperator;
use Rx\Operator\CountOperator;
use Rx\Operator\DefaultIfEmptyOperator;
use Rx\Operator\DeferOperator;
use Rx\Operator\DelayOperator;
use Rx\Operator\DematerializeOperator;
use Rx\Operator\DistinctOperator;
use Rx\Operator\DistinctUntilChangedOperator;
use Rx\Operator\DoOnEachOperator;
use Rx\Operator\GroupByUntilOperator;
use Rx\Operator\MapOperator;
use Rx\Operator\FilterOperator;
use Rx\Operator\MinOperator;
use Rx\Operator\MaxOperator;
use Rx\Operator\MaterializeOperator;
use Rx\Operator\MergeAllOperator;
use Rx\Operator\RaceOperator;
use Rx\Operator\ReduceOperator;
use Rx\Operator\RepeatOperator;
use Rx\Operator\RepeatWhenOperator;
use Rx\Operator\RetryOperator;
use Rx\Operator\RetryWhenOperator;
use Rx\Operator\ScanOperator;
use Rx\Operator\SkipLastOperator;
use Rx\Operator\SkipOperator;
use Rx\Operator\SkipUntilOperator;
use Rx\Operator\StartWithArrayOperator;
use Rx\Operator\SkipWhileOperator;
use Rx\Operator\SubscribeOnOperator;
use Rx\Operator\SwitchFirstOperator;
use Rx\Operator\SwitchLatestOperator;
use Rx\Operator\TakeLastOperator;
use Rx\Operator\TakeOperator;
use Rx\Operator\TakeUntilOperator;
use Rx\Operator\TakeWhileOperator;
use Rx\Operator\ThrottleOperator;
use Rx\Operator\TimeoutOperator;
use Rx\Operator\TimestampOperator;
use Rx\Operator\ToArrayOperator;
use Rx\Operator\ZipOperator;
use Rx\Scheduler\ImmediateScheduler;
use Rx\Subject\AsyncSubject;
use Rx\Subject\BehaviorSubject;
use Rx\Subject\ReplaySubject;
use Rx\Subject\Subject;
use Rx\Disposable\EmptyDisposable;
use Rx\Disposable\CallbackDisposable;

class Observable implements ObservableInterface
{
    protected $observers = [];
    protected $started = false;

    public function subscribe(ObserverInterface $observer, $scheduler = null)
    {
        $this->observers[] = $observer;
        $this->started     = true;

        return new CallbackDisposable(function () use ($observer) {
            $this->removeObserver($observer);
        });
    }

    /**
     * @internal
     */
    public function removeObserver(ObserverInterface $observer)
    {
        $key = array_search($observer, $this->observers, true);

        if (false === $key) {
            return false;
        }

        unset($this->observers[$key]);

        return true;
    }

    public function subscribeCallback(callable $onNext = null, callable  $onError = null, callable $onCompleted = null, SchedulerInterface $scheduler = null)
    {
        $observer = new CallbackObserver($onNext, $onError, $onCompleted);

        return $this->subscribe($observer, $scheduler);
    }

    /**
     * Creates an observable sequence from a specified subscribeAction callable implementation.
     *
     * @param callable $subscribeAction Implementation of the resulting observable sequence's subscribe method.
     * @return AnonymousObservable The observable sequence with the specified implementation for the subscribe method.
     *
     * @demo create/create.php
     * @operator
     * @reactivex create
     */
    public static function create(callable $subscribeAction)
    {
        return new AnonymousObservable($subscribeAction);
    }

    /**
     * Returns an Observable that emits an infinite sequence of ascending integers starting at 0, with a constant interval of time of your choosing between emissions.
     *
     * @param $interval int Period for producing the values in the resulting sequence (specified as an integer denoting milliseconds).
     * @param SchedulerInterface|null $scheduler
     * @return IntervalObservable An observable sequence that produces a value after each period.
     *
     * @demo interval/interval.php
     * @operator
     * @reactivex interval
     */
    public static function interval($interval, $scheduler = null)
    {
        return new IntervalObservable($interval, $scheduler);
    }

    /**
     * Returns an observable sequence that contains a single element.
     *
     * @param mixed $value Single element in the resulting observable sequence.
     * @return \Rx\Observable\AnonymousObservable An observable sequence with the single element.
     *
     * @demo just/just.php
     * @operator
     * @reactivex just
     */
    public static function just($value)
    {
        return new ReturnObservable($value);
    }

    /**
     * Returns an empty observable sequence.
     *
     * @return EmptyObservable An observable sequence with no elements.
     *
     * @demo empty-observable/empty-observable.php
     * @operator
     * @reactivex empty-never-throw
     */
    public static function emptyObservable()
    {
        return new EmptyObservable();
    }

    /**
     * Returns a non-terminating observable sequence, which can be used to denote an infinite duration.
     *
     * @return NeverObservable An observable sequence whose observers will never get called.
     *
     * @demo never/never.php
     * @operator
     * @reactivex empty-never-throw
     */
    public static function never()
    {
        return new NeverObservable();
    }

    /**
     * Returns an observable sequence that terminates with an exception.
     *
     * @param $error
     * @return ErrorObservable The observable sequence that terminates exceptionally with the specified exception object.
     *
     * @demo error-observable/error-observable.php
     * @operator
     * @reactivex empty-never-throw
     */
    public static function error(\Exception $error)
    {
        return new ErrorObservable($error);
    }

    /**
     * Combine an Observable together with another Observable by merging their emissions into a single Observable.
     *
     * @param ObservableInterface $otherObservable
     * @return AnonymousObservable
     *
     * @demo merge/merge.php
     * @operator
     * @reactivex merge
     */
    public function merge(ObservableInterface $otherObservable)
    {
        return self::mergeAll(new AnonymousObservable(function (ObserverInterface $observer, SchedulerInterface $schedule) use ($otherObservable) {
            $observer->onNext($this);
            $observer->onNext($otherObservable);
            $observer->onCompleted();
        }));
    }

    /**
     * Merges an observable sequence of observables into an observable sequence.
     *
     * @param ObservableInterface $sources
     * @return AnonymousObservable
     *
     * @demo merge/merge-all.php
     * @operator
     * @reactivex merge
     */
    public static function mergeAll(ObservableInterface $sources)
    {
        return (new EmptyObservable())->lift(function () use ($sources) {
            return new MergeAllOperator($sources);
        });
    }

    /**
     * Converts an array to an observable sequence
     *
     * @param array $array
     * @return ArrayObservable
     *
     * @demo fromArray/fromArray.php
     * @operator
     * @reactivex from
     */
    public static function fromArray(array $array)
    {
        return new ArrayObservable($array);
    }

    /**
     * Converts an Iterator into an observable sequence
     *
     * @param \Iterator $iterator
     * @return IteratorObservable
     *
     * @demo iterator/iterator.php
     * @operator
     * @reactivex from
     */
    public static function fromIterator(\Iterator $iterator)
    {
        return new IteratorObservable($iterator);
    }

    /**
     * Returns an observable sequence that invokes the specified factory function whenever a new observer subscribes.
     *
     * @param callable $factory
     * @return \Rx\Observable\AnonymousObservable
     *
     * @demo defer/defer.php
     * @operator
     * @reactivex defer
     */
    public static function defer(callable $factory)
    {
        return (new EmptyObservable())->lift(function () use ($factory) {
            return new DeferOperator($factory);
        });
    }

    /**
     * Generates an observable sequence of integral numbers within a specified range, using the specified scheduler to
     * send out observer messages.
     *
     * @param $start
     * @param $count
     * @return RangeObservable
     *
     * @demo range/range.php
     * @operator
     * @reactivex range
     */
    public static function range($start, $count, SchedulerInterface $scheduler = null)
    {
        return new RangeObservable($start, $count, $scheduler);
    }

    /**
     * Invokes the specified function asynchronously on the specified scheduler, surfacing the result through an
     * observable sequence.
     *
     * @param callable $action
     * @param SchedulerInterface $scheduler
     * @return AnonymousObservable
     *
     * @demo start/start.php
     * @operator
     * @reactivex start
     */
    public static function start(callable $action, SchedulerInterface $scheduler = null)
    {
        $scheduler = $scheduler ?: new ImmediateScheduler();
        $subject   = new AsyncSubject();

        $scheduler->schedule(function () use ($subject, $action) {
            $result = null;
            try {
                $result = call_user_func($action);
            } catch (\Exception $e) {
                $subject->onError($e);
                return;
            }
            $subject->onNext($result);
            $subject->onCompleted();
        });

        return $subject->asObservable();
    }

    /**
     * Takes a transforming function that operates on each element.
     *
     * @param callable $selector
     * @return AnonymousObservable
     *
     * @demo map/map.php
     * @operator
     * @reactivex map
     */
    public function map(callable $selector)
    {
        return $this->lift(function () use ($selector) {
            return new MapOperator($selector);
        });
    }

    /**
     * Maps operator variant that calls the map selector with the index and value
     *
     * @param callable $selector
     * @return AnonymousObservable
     *
     * @demo map/mapWithIndex.php
     * @operator
     * @reactivex map
     */
    public function mapWithIndex(callable $selector)
    {
        $index = 0;
        return $this->map(function ($value) use ($selector, &$index) {
            return call_user_func_array($selector, [$index++, $value]);
        });
    }

    /**
     * Maps every value to the same value every time
     *
     * @param $value
     * @return AnonymousObservable
     *
     * @demo map/mapTo.php
     * @operator
     * @reactivex map
     */
    public function mapTo($value)
    {
        return $this->map(function () use ($value) {
            return $value;
        });
    }

    /**
     * Alias for Map
     *
     * @param callable $selector
     * @return \Rx\Observable\AnonymousObservable
     *
     * @operator
     * @reactivex map
     */
    public function select(callable $selector)
    {
        return $this->map($selector);
    }

    /**
     * Emit only those items from an Observable that pass a predicate test.
     *
     * @param callable $predicate
     * @return \Rx\Observable\AnonymousObservable
     *
     * @demo filter/filter.php
     * @operator
     * @reactivex filter
     */
    public function filter(callable $predicate)
    {
        return $this->lift(function () use ($predicate) {
            return new FilterOperator($predicate);
        });
    }

    /**
     * Alias for filter
     *
     * @param callable $predicate
     * @return \Rx\Observable\AnonymousObservable
     *
     * @operator
     * @reactivex filter
     */
    public function where(callable $predicate)
    {
        return $this->filter($predicate);
    }

    /**
     * Projects each element of an observable sequence to an observable sequence and merges the resulting observable sequences into one observable sequence.
     *
     * @param callable $selector
     * @return AnonymousObservable
     *
     * @demo flatMap/flatMap.php
     * @operator
     * @reactivex flatMap
     */
    public function flatMap(callable $selector)
    {
        return self::mergeAll($this->select($selector));
    }

    /**
     * Projects each element of the source observable sequence to the other observable sequence and merges the
     * resulting observable sequences into one observable sequence.
     *
     * @param ObservableInterface $observable - An an observable sequence to project each element from the source
     * sequence onto.
     *
     * @return AnonymousObservable
     *
     * @demo concat/flatMapTo.php
     * @operator
     * @reactivex flatMap
     */
    public function flatMapTo(ObservableInterface $observable)
    {
        return $this->flatMap(function () use ($observable) {
            return $observable;
        });
    }

    /**
     * Alias for flatMap
     *
     * @param $selector
     * @return AnonymousObservable
     *
     * @operator
     * @reactivex flatMap
     */
    public function selectMany($selector)
    {
        return $this->flatMap($selector);
    }

    /**
     * Bypasses a specified number of elements in an observable sequence and then returns the remaining elements.
     *
     * Transform the items emitted by an Observable into Observables, and mirror those items emitted by the
     * most-recently transformed Observable.
     *
     * The flatMapLatest operator is similar to the flatMap and concatMap methods described above, however, rather than
     * emitting all of the items emitted by all of the Observables that the operator generates by transforming items
     * from the source Observable, flatMapLatest instead emits items from each such transformed Observable only until
     * the next such Observable is emitted, then it ignores the previous one and begins emitting items emitted by the
     * new one.
     *
     * @param callable $selector - A transform function to apply to each source element.
     * @return AnonymousObservable - An observable sequence which transforms the items emitted by an Observable into
     * Observables, and mirror those items emitted by the most-recently transformed Observable.
     *
     * @demo flatMap/flatMapLatest.php
     * @operator
     * @reactivex flatMap
     */
    public function flatMapLatest(callable $selector, SchedulerInterface $scheduler = null)
    {
        return $this->map($selector)->switchLatest($scheduler);
    }

    /**
     * @param integer $count
     * @return AnonymousObservable
     *
     * @demo skip/skip.php
     * @operator
     * @reactivex skip
     */
    public function skip($count)
    {
        return $this->lift(function () use ($count) {
            return new SkipOperator($count);
        });
    }

    /**
     * Bypasses elements in an observable sequence as long as a specified condition is true and then returns the
     * remaining elements.
     *
     * @param callable $predicate A function to test each element for a condition.
     *
     * @return AnonymousObservable An observable sequence that contains the elements from the input sequence starting
     * at the first element in the linear series that does not pass the test specified by predicate.
     *
     * @demo skip/skipWhile.php
     * @operator
     * @reactivex skipWhile
     */
    public function skipWhile(callable $predicate)
    {
        return $this->lift(function () use ($predicate) {
            return new SkipWhileOperator($predicate);
        });
    }

    /**
     * Bypasses elements in an observable sequence as long as a specified condition is true and then returns the
     * remaining elements. The element's index is used in the logic of the predicate function.
     *
     * @param callable $predicate A function to test each element for a condition; the first parameter of the
     * function represents the index of the source element, the second parameter is the value.
     *
     * @return AnonymousObservable An observable sequence that contains the elements from the input sequence starting
     * at the first element in the linear series that does not pass the test specified by predicate.
     *
     * @demo skip/skipWhileWithIndex.php
     * @operator
     * @reactivex skipWhile
     */
    public function skipWhileWithIndex(callable $predicate)
    {
        $index = 0;
        return $this->skipWhile(function ($value) use ($predicate, &$index) {
            return call_user_func_array($predicate, [$index++, $value]);
        });
    }

    /**
     * Returns a specified number of contiguous elements from the start of an observable sequence
     *
     * @param integer $count
     * @return AnonymousObservable|EmptyObservable
     *
     * @demo take/take.php
     * @operator
     * @reactivex take
     */
    public function take($count)
    {
        if ($count === 0) {
            return new EmptyObservable();
        }

        return $this->lift(function () use ($count) {
            return new TakeOperator($count);
        });
    }

    /**
     * Returns the values from the source observable sequence until the other observable sequence produces a value.
     *
     * @param ObservableInterface $other - other Observable sequence that terminates propagation of elements of
     * the source sequence.
     * @return AnonymousObservable - An observable sequence containing the elements of the source sequence up to the
     * point the other sequence interrupted further propagation.
     *
     * @demo take/take.php
     * @operator
     * @reactivex take
     */
    public function takeUntil(ObservableInterface $other)
    {
        return $this->lift(function () use ($other) {
            return new TakeUntilOperator($other);
        });
    }

    /**
     * Returns elements from an observable sequence as long as a specified condition is true.  It takes as a parameter a
     * a callback to test each source element for a condition.  The callback predicate is called with the value of the
     * element.
     *
     * @param callable $predicate
     * @return AnonymousObservable
     *
     * @demo take/takeWhile.php
     * @operator
     * @reactivex takeWhile
     */
    public function takeWhile(callable $predicate)
    {
        return $this->lift(function () use ($predicate) {
            return new TakeWhileOperator($predicate);
        });
    }

    /**
     * Returns elements from an observable sequence as long as a specified condition is true.  It takes as a parameter a
     * a callback to test each source element for a condition.  The callback predicate is called with the index and the
     * value of the element.
     *
     * @param callable $predicate
     * @return AnonymousObservable
     *
     * @demo take/takeWhileWithIndex.php
     * @operator
     * @reactivex takeWhile
     */
    public function takeWhileWithIndex(callable $predicate)
    {
        $index = 0;
        return $this->takeWhile(function ($value) use ($predicate, &$index) {
            return call_user_func_array($predicate, [$index++, $value]);
        });
    }

    /**
     * Returns a specified number of contiguous elements from the end of an observable sequence.
     *
     * @param $count
     * @return AnonymousObservable
     *
     * @demo take/takeLast.php
     * @operator
     * @reactivex takeLast
     */
    public function takeLast($count)
    {
        return $this->lift(function () use ($count) {
            return new TakeLastOperator($count);
        });
    }

    /**
     * Groups the elements of an observable sequence according to a specified key selector function and comparer and selects the resulting elements by using a specified function.
     *
     * @param callable $keySelector
     * @param callable|null $elementSelector
     * @param callable|null $keySerializer
     * @return AnonymousObservable
     *
     * @demo groupBy/groupBy.php
     * @operator
     * @reactivex groupBy
     */
    public function groupBy(callable $keySelector, callable $elementSelector = null, callable $keySerializer = null)
    {
        return $this->groupByUntil($keySelector, $elementSelector, function () {

            // observable that never calls
            return new AnonymousObservable(function () {
                // todo?
                return new EmptyDisposable();
            });
        }, $keySerializer);
    }

    /**
     * Groups the elements of an observable sequence according to a specified key selector function and comparer and selects the resulting elements by using a specified function.
     *
     * @param callable $keySelector
     * @param callable|null $elementSelector
     * @param callable|null $durationSelector
     * @param callable|null $keySerializer
     * @return AnonymousObservable
     *
     * @demo groupBy/groupByUntil.php
     * @operator
     * @reactivex groupBy
     */
    public function groupByUntil(callable $keySelector, callable $elementSelector = null, callable $durationSelector = null, callable $keySerializer = null)
    {
        return $this->lift(function () use ($keySelector, $elementSelector, $durationSelector, $keySerializer) {
            return new GroupByUntilOperator($keySelector, $elementSelector, $durationSelector, $keySerializer);
        });
    }

    /**
     * Lifts a function to the current Observable and returns a new Observable that when subscribed to will pass
     * the values of the current Observable through the Operator function.
     *
     * @param callable $operatorFactory
     * @return AnonymousObservable
     */
    public function lift(callable $operatorFactory)
    {
        return new AnonymousObservable(function (ObserverInterface $observer, SchedulerInterface $schedule) use ($operatorFactory) {
            $operator = $operatorFactory();
            return $operator($this, $observer, $schedule);
        });
    }

    /**
     * This method allows the use of extra operators with the namespace:
     * Rx\Operator
     * and also custom operators by adding an operator class with the
     * namespace format:
     * CustomNamespace\Rx\Operator\OperatorNameOperator
     *
     * @param $name
     * @param $arguments
     * @return Observable
     * 
     * @demo custom-operator/rot13.php
     */
    public function __call($name, array $arguments)
    {
        $fullNamespace = 'Rx\\Operator\\';
        if ($name[0] === '_') {
            list($_, $namespace, $methodName) = explode('_', $name);
            $name = $methodName;
{'fraction_non_alphanumeric': 0.0888190001084481, 'fraction_numerical': 0.00041210280880598634, 'mean_word_length': 2.624685534591195, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 77, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2319302', 'n_tokens_mistral': 12502, 'n_tokens_neox': 11818, 'n_words': 3772}