Rx.PHP currently supports PHP 5.3. https://github.com/asm89/Rx.PHP/pull/6 changes the requirement to PHP 5.4, but that version is already EOL (http://php.net/eol.php).

How about raising the minimal supported version to either 5.5 or 5.6 and up?

cc @davidwdan @cboden @mbonneau @Ocramius

The following code will cause php to seg fault:

$observable = new Rx\Observable\ArrayObservable(range(1, 10000));
$observable->subscribe(new \Rx\Observer\CallbackObserver());

I'm guessing that this has something to do with the recursive actions in the ImmediateScheduler blowing the stack, because it works fine with the VirtualTimeScheduler.

Every other implementation of Rx, other than RxNet, is now using map and flatMap. We should probably consider changing it or at least making an alias.

If you subscribe without a scheduler to the AnonymousObservable returned by BaseObservable::fromArray, the script dies.

Fatal error: Call to a member function scheduleRecursive() on null in  rx.php/lib/Rx/Observable/BaseObservable.php on line 223

Also, fromArray is called in BaseObservable::merge with a scheduler argument when it doesn't take an argument.

Example code:

Failure on merge:

require_once __DIR__ . '/vendor/autoload.php';

$aa = [1,2,3,4];
$ab = [5,6,7,8];

$oa = new \Rx\Observable\ArrayObservable($aa);
$ob = new \Rx\Observable\ArrayObservable($ab);

$oa->merge($ob)->subscribeCallback(function ($x) {
    echo "$x\n";
});

Failure with fromArray:

require_once __DIR__ . '/vendor/autoload.php';

$aa = [1,2,3,4];

$oa = \Rx\Observable\BaseObservable::fromArray($aa);

$oa->subscribeCallback(function ($x) {
    echo "$x\n";
});

PHP 5.5 is the oldest still supported PHP version. It does not make sense to support older versions. PHP 7 will be released soon, so let's start explicitly supporting it.

Fixes #7

With the current version of Rx.PHP, if you want to implement a custom operator, you need to add it to BaseObservable.

I propose that we take a page from RxJava and add a lift operator, which would allow custom operators to be added to the chain.

For Example:

$observable = new Rx\Observable\ArrayObservable(array(21, 21, 42));
$observable
    ->select(function($elem) { return $elem * 2; })
    ->lift(new DistinctUntilChangedOperator())
    ->subscribe($stdoutObserver);

I've been working on an implementation of this here: https://github.com/davidwdan/Rx.PHP/compare/master...operators?diff=unified&name=operators

It includes the lift operator along with tap, never and distinctUntilChanged.

If this is a direction you're interested in going, let me know and I'll create PRs.

ArrayObservable uses array_shift instead of an index to access the current value, which makes it about 10 times slower than fromArray

I can't think of any reasons why we should have both implementations. The static method should return a new instance of ArrayObservable, which should use the index instead of array_shift.

Another issue, is that fromArray only works with a numeric index. If you use an associative array, it fails.

We should use array_keys to get the actual keys, and then do this: $observer->onNext($array[$keys[$count]]);

The key will be lost, which will be okay in most cases. If you need to preserve the keys, you can use the pairs operator.

I would expect the output of the code below:

$outputObserver = new \Rx\Observer\CallbackObserver(function ($x) { echo $x . "\n"; });
$outputObserver2 = new \Rx\Observer\CallbackObserver(function ($x) { echo $x . "\n"; });

$a = [1,2,3,4,5];

$o = \Rx\Observable\BaseObservable::fromArray($a);

$count = $o->count();

$count->subscribe($outputObserver);

$count->subscribe($outputObserver2);

to be:

5
5

but it actually outputs:

5
10

Because there is only one instance of the operator created, the same instance is being used for both subscriptions and it counts twice.

To fix this, the argument to lift should probably be an operator factory so every subscription gets a different instance.

And lift would change to:

    public function lift(callable $operatorFactory)
    {
        return new AnonymousObservable(function (ObserverInterface $observer, SchedulerInterface $schedule) use ($operatorFactory) {
            $operator = $operatorFactory();
            return $operator($this, $observer, $schedule);
        });
    }

I've implemented all of the connectable observables from RxJS. Since they're interdependent, it probably makes sense to handle them as one PR, but I can split them up if it's too much to review at once.

Before I create the PR, I need some feedback on the best way to handle the operators that can take a value or factory as the same argument. See:

https://github.com/davidwdan/Rx.PHP/blob/connectable_observables/lib/Rx/Observable/BaseObservable.php#L650 https://github.com/davidwdan/Rx.PHP/blob/connectable_observables/lib/Rx/Observable/BaseObservable.php#L697

Also, some of the tests are skipped because zip hasn't been implemented yet. @JMB3 is working on that.