more benchmarks

benchmark/catch/catch.php

@@ -0,0 +1,12 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::error(new \Exception('error'))
+    ->catchError(function() {
+        return Observable::just(25);
+    });
+
+return function() use ($source) {
+    return $source;
+};

benchmark/combineLatest/combineLatest.php

@@ -0,0 +1,12 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(0, 25)
+    ->combineLatest([Observable::range(0, 25)], function($a, $b) {
+        return $a + $b;
+    });
+
+return function() use ($source) {
+    return $source;
+};

benchmark/combineLatest/combineLatest_eventloop.php

@@ -0,0 +1,19 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\EventLoopScheduler;
+use React\EventLoop\StreamSelectLoop;
+
+$loop = new StreamSelectLoop();
+$scheduler = new EventLoopScheduler($loop);
+
+$source = Observable::range(0, 25, $scheduler)
+    ->combineLatest([Observable::range(0, 25, $scheduler)], function($a, $b) {
+        return $a + $b;
+    });
+
+$factory =  function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];

benchmark/concat/concat.php

@@ -2,8 +2,8 @@
 
 use Rx\Observable;
 
-$source = Observable::range(1, 25)
-    ->concat(Observable::range(1, 25));
+$source = Observable::range(0, 25)
+    ->concat(Observable::range(0, 25));
 
 return function() use ($source) {
     return $source;

benchmark/concat/concat_eventloop.php

@@ -7,8 +7,8 @@
 $loop = new StreamSelectLoop();
 $scheduler = new EventLoopScheduler($loop);
 
-$source = Observable::range(1, 25, $scheduler)
-    ->concat(Observable::range(1, 25, $scheduler));
+$source = Observable::range(0, 25, $scheduler)
+    ->concat(Observable::range(0, 25, $scheduler));
 
 $factory = function() use ($source) {
     return $source;

benchmark/defaultIfEmpty/defaultIfEmpty.php

@@ -0,0 +1,10 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::emptyObservable()
+    ->defaultIfEmpty(Observable::just(25));
+
+return function() use ($source) {
+    return $source;
+};

benchmark/distinct/distinct.php

@@ -2,11 +2,10 @@
 
 use Rx\Observable;
 
-$range = array_map(function($val) {
-    return $val % 3;
-}, range(0, 25));
-
-$source = Observable::fromArray($range)
+$source = Observable::range(0, 25)
+    ->map(function($i) {
+        return $i % 3;
+    })
     ->distinct();
 
 return function() use ($source) {

benchmark/distinct/distinct_eventloop.php

@@ -8,11 +8,10 @@
 $loop = new StreamSelectLoop();
 $scheduler = new EventLoopScheduler($loop);
 
-$range = array_map(function($val) {
-    return $val % 3;
-}, range(0, 25));
-
-$source = (new ArrayObservable($range, $scheduler))
+$source = Observable::range(0, 25, $scheduler)
+    ->map(function($i) {
+        return $i % 3;
+    })
     ->distinct();
 
 $factory = function() use ($source) {

benchmark/forkJoin/forkJoin.php

@@ -0,0 +1,15 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::defer(function() {
+    return Observable::forkJoin([
+        Observable::just(25),
+        Observable::range(0, 25),
+        Observable::fromArray(([1, 2, 3, 4, 5]))
+    ]);
+});
+
+return function() use ($source) {
+    return $source;
+};

benchmark/groupBy/groupBy.php

@@ -0,0 +1,15 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(0, 25)
+    ->map(function($i) {
+        return ['key' => $i % 5];
+    })
+    ->groupBy(function($item) {
+        return $item['key'];
+    });
+
+return function() use ($source) {
+    return $source;
+};

benchmark/groupBy/groupBy_eventloop.php

@@ -0,0 +1,22 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\EventLoopScheduler;
+use React\EventLoop\StreamSelectLoop;
+
+$loop = new StreamSelectLoop();
+$scheduler = new EventLoopScheduler($loop);
+
+$source = Observable::range(0, 25, $scheduler)
+    ->map(function($i) {
+        return ['key' => $i % 5];
+    })
+    ->groupBy(function($item) {
+        return $item['key'];
+    });
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];

benchmark/isEmpty/isEmpty.php

@@ -0,0 +1,10 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::just(25)
+    ->isEmpty();
+
+return function() use ($source) {
+    return $source;
+};

benchmark/merge/merge.php

@@ -2,8 +2,8 @@
 
 use Rx\Observable;
 
-$source = Observable::range(1, 25)
-    ->merge(Observable::range(1, 25));
+$source = Observable::range(0, 25)
+    ->merge(Observable::range(0, 25));
 
 return function() use ($source) {
     return $source;

benchmark/merge/merge_eventloop.php

@@ -7,8 +7,8 @@
 $loop = new StreamSelectLoop();
 $scheduler = new EventLoopScheduler($loop);
 
-$source = Observable::range(0, 250, $scheduler)
-    ->merge(Observable::range(0, 250, $scheduler));
+$source = Observable::range(0, 25, $scheduler)
+    ->merge(Observable::range(0, 25, $scheduler));
 
 $factory = function() use ($source) {
     return $source;

benchmark/mergeAll/mergeAll_eventloop.php

@@ -7,9 +7,9 @@
 $loop = new StreamSelectLoop();
 $scheduler = new EventLoopScheduler($loop);
 
-$source = Observable::range(0, 25)
-    ->map(function() {
-        return Observable::range(0, 25);
+$source = Observable::range(0, 25, $scheduler)
+    ->map(function() use ($scheduler) {
+        return Observable::range(0, 25, $scheduler);
     })
     ->mergeAll();
 

benchmark/pluck/pluck.php

@@ -0,0 +1,10 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::just([1, 2, 3, 4, 5])
+    ->pluck(2);
+
+return function() use ($source) {
+    return $source;
+};

benchmark/reduce/reduce.php

@@ -0,0 +1,12 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(0, 25)
+    ->reduce(function($a, $b) {
+        return $a + $b;
+    });
+
+return function() use ($source) {
+    return $source;
+};

benchmark/reduce/reduce_eventloop.php

@@ -0,0 +1,19 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\EventLoopScheduler;
+use React\EventLoop\StreamSelectLoop;
+
+$loop = new StreamSelectLoop();
+$scheduler = new EventLoopScheduler($loop);
+
+$source = Observable::range(0, 25, $scheduler)
+    ->reduce(function($a, $b) {
+        return $a + $b;
+    });
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];

benchmark/repeat/repeat.php

@@ -0,0 +1,10 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(0, 25)
+    ->repeat(5);
+
+return function() use ($source) {
+    return $source;
+};

benchmark/repeat/repeat_eventloop.php

@@ -0,0 +1,17 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\EventLoopScheduler;
+use React\EventLoop\StreamSelectLoop;
+
+$loop = new StreamSelectLoop();
+$scheduler = new EventLoopScheduler($loop);
+
+$source = Observable::range(0, 25, $scheduler)
+    ->repeat(5);
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];

benchmark/retry/retry.php

@@ -0,0 +1,19 @@
+<?php
+
+use Rx\Observable;
+
+$maxRetryCount = 25;
+$newRetryCount = 0;
+
+$source = Observable::range(5, 1)
+    ->flatMap(function($x) use (&$maxRetryCount, &$newRetryCount) {
+        if (++$newRetryCount < $maxRetryCount - 1) {
+            return Observable::error(new \Exception('error'));
+        }
+        return Observable::just($x);
+    })
+    ->retry($maxRetryCount);
+
+return function() use ($source) {
+    return $source;
+};

benchmark/retry/retry_eventloop.php

@@ -0,0 +1,26 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\EventLoopScheduler;
+use React\EventLoop\StreamSelectLoop;
+
+$loop = new StreamSelectLoop();
+$scheduler = new EventLoopScheduler($loop);
+
+$maxRetryCount = 25;
+$newRetryCount = 0;
+
+$source = Observable::range(5, 1, $scheduler)
+    ->flatMap(function($x) use (&$maxRetryCount, &$newRetryCount) {
+        if (++$newRetryCount < $maxRetryCount - 1) {
+            return Observable::error(new \Exception('error'));
+        }
+        return Observable::just($x);
+    })
+    ->retry($maxRetryCount);
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];

benchmark/scan/scan.php

@@ -0,0 +1,12 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(0, 25)
+    ->scan(function($acc, $x) {
+        return $x + $x;
+    });
+
+return function() use ($source) {
+    return $source;
+};

benchmark/scan/scan_eventloop.php

@@ -0,0 +1,19 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\EventLoopScheduler;
+use React\EventLoop\StreamSelectLoop;
+
+$loop = new StreamSelectLoop();
+$scheduler = new EventLoopScheduler($loop);
+
+$source = Observable::range(0, 25, $scheduler)
+    ->scan(function($acc, $x) {
+        return $x + $x;
+    });
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];

benchmark/skip/skip.php

@@ -0,0 +1,10 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(0, 50)
+    ->skip(25);
+
+return function() use ($source) {
+    return $source;
+};