From 4dd160a94c56b8832053b7f7be4e73f83d7157cb Mon Sep 17 00:00:00 2001
From: Martin Sikora <martin.sikora.ahoj@gmail.com>
Date: Sun, 5 Feb 2017 16:03:26 +0100
Subject: [PATCH 01/15] benchmark runner

---
 benchmark/run.php | 105 ++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 105 insertions(+)
 create mode 100644 benchmark/run.php

diff --git a/benchmark/run.php b/benchmark/run.php
new file mode 100644
index 00000000..393dc7fe
--- /dev/null
+++ b/benchmark/run.php
@@ -0,0 +1,105 @@
+<?php
+
+if (file_exists($file = __DIR__.'/../vendor/autoload.php')) {
+    $autoload = require_once $file;
+    $autoload->addPsr4('Vendor\\Rx\\Operator\\', __DIR__ . '/custom-operator');
+} else {
+    throw new RuntimeException('Install dependencies to run benchmark suite.');
+}
+
+use Rx\Observable;
+use Rx\Observer\CallbackObserver;
+
+define('MIN_TOTAL_DURATION', 5);
+$start = microtime(true);
+
+if ($_SERVER['argc'] == 1) {
+    $files = glob(__DIR__ . '/**/*.php');
+} else {
+    // Force absolute path
+    $files = array_map(function($file) {
+        return $file[0] == DIRECTORY_SEPARATOR ? $file : $_SERVER['PWD'] . DIRECTORY_SEPARATOR . $file;
+    }, array_slice($_SERVER['argv'], 1));
+}
+
+Observable::just($files)
+    ->doOnNext(function(array $files) {
+        printf("Benchmarking %d file/s (min %ds each)\n", count($files), MIN_TOTAL_DURATION);
+        printf("script_name - total_runs (single_run_mean ±standard_variant)\n");
+        printf("============================================================\n");
+    })
+    ->concatMap(function($files) { // flatten the array
+        return Observable::fromArray($files);
+    })
+    ->doOnNext(function($file) {
+        printf('%s', pathinfo($file, PATHINFO_FILENAME));
+    })
+    ->map(function($file) { // run benchmark
+        $totalDuration = 0.0;
+        $durations = [];
+
+        ob_start();
+
+        $dummyObserver = new Rx\Observer\CallbackObserver(
+            function ($value) { },
+            function ($error) { },
+            function () { }
+        );
+
+        $testClosure = @include $file;
+        if (!$testClosure) {
+            throw new Exception("Unable to load file \"$file\"");
+        }
+
+        while ($totalDuration < MIN_TOTAL_DURATION) {
+            $start = microtime(true);
+
+            $testClosure();
+
+            $duration = microtime(true) - $start;
+
+            $durations[] = $duration * 1000;
+            $totalDuration += $duration;
+        }
+
+        ob_end_clean();
+
+        return [
+            'file' => $file,
+            'durations' => $durations,
+        ];
+    })
+    ->doOnNext(function($result) { // Print the number of successful runs
+        printf(' - %d', count($result['durations']));
+    })
+    ->map(function($result) { // Calculate the standard deviance
+        $mean = array_sum($result['durations']) / count($result['durations']);
+
+        $variance = array_sum(array_map(function($duration) use ($mean) {
+            return pow($mean - $duration, 2);
+        }, $result['durations']));
+
+        return [
+            'file' => $result['file'],
+            'mean' => $mean,
+            'standard_variance' => pow($variance, 0.5),
+        ];
+    })
+    ->subscribe(new CallbackObserver(
+        function($result) {
+            printf(" (%.2fms ±%.2fms)\n", $result['mean'], $result['standard_variance']);
+        },
+        function(\Exception $error) {
+            printf("\nError: %s\n", $error->getMessage());
+        },
+        function() use ($start) {
+            printf("============================================================\n");
+            printf("total duration: %.2fs\n", microtime(true) - $start);
+        }
+    ));
+
+
+//$dirIter = new RecursiveDirectoryIterator($dir);
+//$iter = new RecursiveIteratorIterator($dirIter);
+//$iter = new DirectoryIterator($dir);
+//$iter = new RegexIterator($iter, $pattern, RegexIterator::GET_MATCH);

From c9466a49d874f0b6de27bae40ead2c21a869c745 Mon Sep 17 00:00:00 2001
From: Martin Sikora <martin.sikora.ahoj@gmail.com>
Date: Sun, 5 Feb 2017 16:03:56 +0100
Subject: [PATCH 02/15] distinct operator benchmark

---
 benchmark/distinct/distinct.php | 9 +++++++++
 1 file changed, 9 insertions(+)
 create mode 100644 benchmark/distinct/distinct.php

diff --git a/benchmark/distinct/distinct.php b/benchmark/distinct/distinct.php
new file mode 100644
index 00000000..c8af7d69
--- /dev/null
+++ b/benchmark/distinct/distinct.php
@@ -0,0 +1,9 @@
+<?php
+
+use Rx\Observable;
+
+return function() use ($dummyObserver) {
+    Observable::range(1, pow(10, 3))
+        ->distinct()
+        ->subscribe($dummyObserver);
+};
\ No newline at end of file

From 560a5ef25dbdb56d1f9dea60fc8c34a4e14d3a4f Mon Sep 17 00:00:00 2001
From: Martin Sikora <martin.sikora.ahoj@gmail.com>
Date: Sun, 5 Feb 2017 16:12:52 +0100
Subject: [PATCH 03/15] fix variant

---
 benchmark/run.php | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/benchmark/run.php b/benchmark/run.php
index 393dc7fe..8664469f 100644
--- a/benchmark/run.php
+++ b/benchmark/run.php
@@ -73,7 +73,8 @@ function () { }
         printf(' - %d', count($result['durations']));
     })
     ->map(function($result) { // Calculate the standard deviance
-        $mean = array_sum($result['durations']) / count($result['durations']);
+        $count = count($result['durations']);
+        $mean = array_sum($result['durations']) / $count;
 
         $variance = array_sum(array_map(function($duration) use ($mean) {
             return pow($mean - $duration, 2);
@@ -82,7 +83,7 @@ function () { }
         return [
             'file' => $result['file'],
             'mean' => $mean,
-            'standard_variance' => pow($variance, 0.5),
+            'standard_variance' => pow($variance / $count, 0.5),
         ];
     })
     ->subscribe(new CallbackObserver(

From 3da18023f01f5572ffec7d8f4108090b0d4f403e Mon Sep 17 00:00:00 2001
From: Martin Sikora <martin.sikora.ahoj@gmail.com>
Date: Sun, 5 Feb 2017 16:14:00 +0100
Subject: [PATCH 04/15] benchmark filter operator

---
 benchmark/filter/filter.php | 11 +++++++++++
 1 file changed, 11 insertions(+)
 create mode 100644 benchmark/filter/filter.php

diff --git a/benchmark/filter/filter.php b/benchmark/filter/filter.php
new file mode 100644
index 00000000..630905f2
--- /dev/null
+++ b/benchmark/filter/filter.php
@@ -0,0 +1,11 @@
+<?php
+
+use Rx\Observable;
+
+return function() use ($dummyObserver) {
+    Observable::range(1, pow(10, 3))
+        ->filter(function($value) {
+            return $value % 2 == 0;
+        })
+        ->subscribe($dummyObserver);
+};
\ No newline at end of file

From 58e654aba07b1002c10f325f31a1272f1101f47d Mon Sep 17 00:00:00 2001
From: Martin Sikora <martin.sikora.ahoj@gmail.com>
Date: Sun, 5 Feb 2017 16:32:19 +0100
Subject: [PATCH 05/15] add types

---
 benchmark/run.php | 17 ++++++-----------
 1 file changed, 6 insertions(+), 11 deletions(-)

diff --git a/benchmark/run.php b/benchmark/run.php
index 8664469f..891411c9 100644
--- a/benchmark/run.php
+++ b/benchmark/run.php
@@ -22,19 +22,20 @@
     }, array_slice($_SERVER['argv'], 1));
 }
 
+
 Observable::just($files)
     ->doOnNext(function(array $files) {
         printf("Benchmarking %d file/s (min %ds each)\n", count($files), MIN_TOTAL_DURATION);
         printf("script_name - total_runs (single_run_mean ±standard_variant)\n");
         printf("============================================================\n");
     })
-    ->concatMap(function($files) { // flatten the array
+    ->concatMap(function($files) { // Flatten the array
         return Observable::fromArray($files);
     })
     ->doOnNext(function($file) {
         printf('%s', pathinfo($file, PATHINFO_FILENAME));
     })
-    ->map(function($file) { // run benchmark
+    ->map(function($file) { // Run benchmark
         $totalDuration = 0.0;
         $durations = [];
 
@@ -69,10 +70,10 @@ function () { }
             'durations' => $durations,
         ];
     })
-    ->doOnNext(function($result) { // Print the number of successful runs
+    ->doOnNext(function(array $result) { // Print the number of successful runs
         printf(' - %d', count($result['durations']));
     })
-    ->map(function($result) { // Calculate the standard deviance
+    ->map(function(array $result) { // Calculate the standard deviance
         $count = count($result['durations']);
         $mean = array_sum($result['durations']) / $count;
 
@@ -87,7 +88,7 @@ function () { }
         ];
     })
     ->subscribe(new CallbackObserver(
-        function($result) {
+        function(array $result) {
             printf(" (%.2fms ±%.2fms)\n", $result['mean'], $result['standard_variance']);
         },
         function(\Exception $error) {
@@ -98,9 +99,3 @@ function() use ($start) {
             printf("total duration: %.2fs\n", microtime(true) - $start);
         }
     ));
-
-
-//$dirIter = new RecursiveDirectoryIterator($dir);
-//$iter = new RecursiveIteratorIterator($dirIter);
-//$iter = new DirectoryIterator($dir);
-//$iter = new RegexIterator($iter, $pattern, RegexIterator::GET_MATCH);

From 8f9879d1ca51f3798f6623fe46cea560b772d80f Mon Sep 17 00:00:00 2001
From: Martin Sikora <martin.sikora.ahoj@gmail.com>
Date: Wed, 8 Feb 2017 19:48:29 +0100
Subject: [PATCH 06/15] added more benchmarks, updated existing from RxJS 5

---
 benchmark/bufferWithCount/bufferWithCount.php |  9 +++++++++
 benchmark/distinct/distinct.php               |  8 ++++++--
 benchmark/filter/filter.php                   |  5 ++++-
 benchmark/skipLast/skipLast.php               |  9 +++++++++
 benchmark/takeLast/takeLast.php               |  9 +++++++++
 benchmark/zip/zip.php                         | 11 +++++++++++
 6 files changed, 48 insertions(+), 3 deletions(-)
 create mode 100644 benchmark/bufferWithCount/bufferWithCount.php
 create mode 100644 benchmark/skipLast/skipLast.php
 create mode 100644 benchmark/takeLast/takeLast.php
 create mode 100644 benchmark/zip/zip.php

diff --git a/benchmark/bufferWithCount/bufferWithCount.php b/benchmark/bufferWithCount/bufferWithCount.php
new file mode 100644
index 00000000..acb4ca4b
--- /dev/null
+++ b/benchmark/bufferWithCount/bufferWithCount.php
@@ -0,0 +1,9 @@
+<?php
+
+use Rx\Observable;
+
+return function() use ($dummyObserver) {
+    Observable::range(0, 25)
+        ->bufferWithCount(5)
+        ->subscribe($dummyObserver);
+};
\ No newline at end of file
diff --git a/benchmark/distinct/distinct.php b/benchmark/distinct/distinct.php
index c8af7d69..120efca2 100644
--- a/benchmark/distinct/distinct.php
+++ b/benchmark/distinct/distinct.php
@@ -2,8 +2,12 @@
 
 use Rx\Observable;
 
-return function() use ($dummyObserver) {
-    Observable::range(1, pow(10, 3))
+$source = array_map(function($val) {
+    return $val % 3;
+}, range(0, 25));
+
+return function() use ($source, $dummyObserver) {
+    Observable::fromArray($source)
         ->distinct()
         ->subscribe($dummyObserver);
 };
\ No newline at end of file
diff --git a/benchmark/filter/filter.php b/benchmark/filter/filter.php
index 630905f2..cb352d06 100644
--- a/benchmark/filter/filter.php
+++ b/benchmark/filter/filter.php
@@ -3,9 +3,12 @@
 use Rx\Observable;
 
 return function() use ($dummyObserver) {
-    Observable::range(1, pow(10, 3))
+    Observable::range(0, 50)
         ->filter(function($value) {
             return $value % 2 == 0;
         })
+        ->filter(function($value) {
+            return $value % 10 == 0;
+        })
         ->subscribe($dummyObserver);
 };
\ No newline at end of file
diff --git a/benchmark/skipLast/skipLast.php b/benchmark/skipLast/skipLast.php
new file mode 100644
index 00000000..3250592f
--- /dev/null
+++ b/benchmark/skipLast/skipLast.php
@@ -0,0 +1,9 @@
+<?php
+
+use Rx\Observable;
+
+return function() use ($dummyObserver) {
+    Observable::range(0, 500)
+        ->skipLast(50)
+        ->subscribe($dummyObserver);
+};
\ No newline at end of file
diff --git a/benchmark/takeLast/takeLast.php b/benchmark/takeLast/takeLast.php
new file mode 100644
index 00000000..c7a2650e
--- /dev/null
+++ b/benchmark/takeLast/takeLast.php
@@ -0,0 +1,9 @@
+<?php
+
+use Rx\Observable;
+
+return function() use ($dummyObserver) {
+    Observable::range(0, 500)
+        ->takeLast(50)
+        ->subscribe($dummyObserver);
+};
\ No newline at end of file
diff --git a/benchmark/zip/zip.php b/benchmark/zip/zip.php
new file mode 100644
index 00000000..f8d11898
--- /dev/null
+++ b/benchmark/zip/zip.php
@@ -0,0 +1,11 @@
+<?php
+
+use Rx\Observable;
+
+return function() use ($dummyObserver) {
+    Observable::range(0, 25)
+        ->zip([Observable::range(0, 25)], function($a, $b) {
+            return $a + $b;
+        })
+        ->subscribe($dummyObserver);
+};
\ No newline at end of file

From 9a06bc2dedb6be298df7377d39b42f73f658e9f0 Mon Sep 17 00:00:00 2001
From: Martin Sikora <martin.sikora.ahoj@gmail.com>
Date: Thu, 9 Feb 2017 20:58:12 +0100
Subject: [PATCH 07/15] added tests using EventLoopScheduler

---
 .../bufferWithCount_eventloop.php             | 16 ++++++++++++
 benchmark/distinct/distinct_eventloop.php     | 20 +++++++++++++++
 benchmark/filter/filter_eventloop.php         | 21 ++++++++++++++++
 benchmark/run.php                             | 25 +++++++++++--------
 benchmark/skipLast/skipLast_eventloop.php     | 16 ++++++++++++
 benchmark/takeLast/takeLast_eventloop.php     | 16 ++++++++++++
 benchmark/zip/zip_eventloop.php               | 18 +++++++++++++
 7 files changed, 122 insertions(+), 10 deletions(-)
 create mode 100644 benchmark/bufferWithCount/bufferWithCount_eventloop.php
 create mode 100644 benchmark/distinct/distinct_eventloop.php
 create mode 100644 benchmark/filter/filter_eventloop.php
 create mode 100644 benchmark/skipLast/skipLast_eventloop.php
 create mode 100644 benchmark/takeLast/takeLast_eventloop.php
 create mode 100644 benchmark/zip/zip_eventloop.php

diff --git a/benchmark/bufferWithCount/bufferWithCount_eventloop.php b/benchmark/bufferWithCount/bufferWithCount_eventloop.php
new file mode 100644
index 00000000..84cdcbd6
--- /dev/null
+++ b/benchmark/bufferWithCount/bufferWithCount_eventloop.php
@@ -0,0 +1,16 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\EventLoopScheduler;
+use React\EventLoop\StreamSelectLoop;
+
+$loop = new StreamSelectLoop();
+$scheduler = new EventLoopScheduler($loop);
+
+return function() use ($dummyObserver, $scheduler, $loop) {
+    Observable::range(0, 25)
+        ->bufferWithCount(5)
+        ->subscribe($dummyObserver, $scheduler);
+
+    $loop->run();
+};
\ No newline at end of file
diff --git a/benchmark/distinct/distinct_eventloop.php b/benchmark/distinct/distinct_eventloop.php
new file mode 100644
index 00000000..703e7c97
--- /dev/null
+++ b/benchmark/distinct/distinct_eventloop.php
@@ -0,0 +1,20 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\EventLoopScheduler;
+use React\EventLoop\StreamSelectLoop;
+
+$loop = new StreamSelectLoop();
+$scheduler = new EventLoopScheduler($loop);
+
+$source = array_map(function($val) {
+    return $val % 3;
+}, range(0, 25));
+
+return function() use ($source, $dummyObserver, $scheduler, $loop) {
+    Observable::fromArray($source)
+        ->distinct()
+        ->subscribe($dummyObserver, $scheduler);
+
+    $loop->run();
+};
\ No newline at end of file
diff --git a/benchmark/filter/filter_eventloop.php b/benchmark/filter/filter_eventloop.php
new file mode 100644
index 00000000..821d406e
--- /dev/null
+++ b/benchmark/filter/filter_eventloop.php
@@ -0,0 +1,21 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\EventLoopScheduler;
+use React\EventLoop\StreamSelectLoop;
+
+$loop = new StreamSelectLoop();
+$scheduler = new EventLoopScheduler($loop);
+
+return function() use ($dummyObserver, $scheduler, $loop) {
+    Observable::range(0, 50)
+        ->filter(function($value) {
+            return $value % 2 == 0;
+        })
+        ->filter(function($value) {
+            return $value % 10 == 0;
+        })
+        ->subscribe($dummyObserver, $scheduler);
+
+    $loop->run();
+};
\ No newline at end of file
diff --git a/benchmark/run.php b/benchmark/run.php
index 891411c9..840c4bad 100644
--- a/benchmark/run.php
+++ b/benchmark/run.php
@@ -13,21 +13,26 @@
 define('MIN_TOTAL_DURATION', 5);
 $start = microtime(true);
 
-if ($_SERVER['argc'] == 1) {
+if ($_SERVER['argc'] === 1) {
     $files = glob(__DIR__ . '/**/*.php');
 } else {
-    // Force absolute path
-    $files = array_map(function($file) {
-        return $file[0] == DIRECTORY_SEPARATOR ? $file : $_SERVER['PWD'] . DIRECTORY_SEPARATOR . $file;
-    }, array_slice($_SERVER['argv'], 1));
+    $files = [];
+    foreach (array_slice($_SERVER['argv'], 1) as $fileOrDir) {
+        if (is_dir($fileOrDir)) {
+            $files = array_merge($files, glob($fileOrDir . '/*.php'));
+        } else {
+            // Force absolute path
+            $files[] = $file[0] === DIRECTORY_SEPARATOR ? $file : $_SERVER['PWD'] . DIRECTORY_SEPARATOR . $file;
+        }
+    }
 }
 
 
 Observable::just($files)
     ->doOnNext(function(array $files) {
         printf("Benchmarking %d file/s (min %ds each)\n", count($files), MIN_TOTAL_DURATION);
-        printf("script_name - total_runs (single_run_mean ±standard_variant)\n");
-        printf("============================================================\n");
+        printf("script_name - total_runs (single_run_mean ±standard_deviation)\n");
+        printf("==============================================================\n");
     })
     ->concatMap(function($files) { // Flatten the array
         return Observable::fromArray($files);
@@ -73,7 +78,7 @@ function () { }
     ->doOnNext(function(array $result) { // Print the number of successful runs
         printf(' - %d', count($result['durations']));
     })
-    ->map(function(array $result) { // Calculate the standard deviance
+    ->map(function(array $result) { // Calculate the standard deviation
         $count = count($result['durations']);
         $mean = array_sum($result['durations']) / $count;
 
@@ -84,12 +89,12 @@ function () { }
         return [
             'file' => $result['file'],
             'mean' => $mean,
-            'standard_variance' => pow($variance / $count, 0.5),
+            'standard_deviation' => pow($variance / $count, 0.5),
         ];
     })
     ->subscribe(new CallbackObserver(
         function(array $result) {
-            printf(" (%.2fms ±%.2fms)\n", $result['mean'], $result['standard_variance']);
+            printf(" (%.2fms ±%.2fms)\n", $result['mean'], $result['standard_deviation']);
         },
         function(\Exception $error) {
             printf("\nError: %s\n", $error->getMessage());
diff --git a/benchmark/skipLast/skipLast_eventloop.php b/benchmark/skipLast/skipLast_eventloop.php
new file mode 100644
index 00000000..e1db6194
--- /dev/null
+++ b/benchmark/skipLast/skipLast_eventloop.php
@@ -0,0 +1,16 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\EventLoopScheduler;
+use React\EventLoop\StreamSelectLoop;
+
+$loop = new StreamSelectLoop();
+$scheduler = new EventLoopScheduler($loop);
+
+return function() use ($dummyObserver, $scheduler, $loop) {
+    Observable::range(0, 500)
+        ->skipLast(50)
+        ->subscribe($dummyObserver, $scheduler);
+
+    $loop->run();
+};
\ No newline at end of file
diff --git a/benchmark/takeLast/takeLast_eventloop.php b/benchmark/takeLast/takeLast_eventloop.php
new file mode 100644
index 00000000..3954ea58
--- /dev/null
+++ b/benchmark/takeLast/takeLast_eventloop.php
@@ -0,0 +1,16 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\EventLoopScheduler;
+use React\EventLoop\StreamSelectLoop;
+
+$loop = new StreamSelectLoop();
+$scheduler = new EventLoopScheduler($loop);
+
+return function() use ($dummyObserver, $scheduler, $loop) {
+    Observable::range(0, 500)
+        ->takeLast(50)
+        ->subscribe($dummyObserver, $scheduler);
+
+    $loop->run();
+};
\ No newline at end of file
diff --git a/benchmark/zip/zip_eventloop.php b/benchmark/zip/zip_eventloop.php
new file mode 100644
index 00000000..6010b631
--- /dev/null
+++ b/benchmark/zip/zip_eventloop.php
@@ -0,0 +1,18 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\EventLoopScheduler;
+use React\EventLoop\StreamSelectLoop;
+
+$loop = new StreamSelectLoop();
+$scheduler = new EventLoopScheduler($loop);
+
+return function() use ($dummyObserver, $scheduler, $loop) {
+    Observable::range(0, 25)
+        ->zip([Observable::range(0, 25)], function($a, $b) {
+            return $a + $b;
+        })
+        ->subscribe($dummyObserver, $scheduler);
+
+    $loop->run();
+};
\ No newline at end of file

From 83ea0cf1bf77d40eefe29b919ad85c93b79f11bc Mon Sep 17 00:00:00 2001
From: Martin Sikora <martin.sikora.ahoj@gmail.com>
Date: Fri, 17 Feb 2017 23:42:28 +0100
Subject: [PATCH 08/15] disallow running benchmarks with xdebug enabled

---
 benchmark/run.php | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/benchmark/run.php b/benchmark/run.php
index 840c4bad..342a31f7 100644
--- a/benchmark/run.php
+++ b/benchmark/run.php
@@ -10,6 +10,12 @@
 use Rx\Observable;
 use Rx\Observer\CallbackObserver;
 
+// Check whether XDebug is enabled
+if (in_array('Xdebug', get_loaded_extensions(true))) {
+    printf("Please, disable Xdebug extension before running RxPHP benchmarks.\n");
+    exit(1);
+}
+
 define('MIN_TOTAL_DURATION', 5);
 $start = microtime(true);
 

From d6365d2d12ced01c683a24aebd51347a280047b9 Mon Sep 17 00:00:00 2001
From: Martin Sikora <martin.sikora.ahoj@gmail.com>
Date: Tue, 21 Feb 2017 22:11:07 +0100
Subject: [PATCH 09/15] log memory usage

---
 benchmark/run.php | 20 +++++++++++++++++---
 1 file changed, 17 insertions(+), 3 deletions(-)

diff --git a/benchmark/run.php b/benchmark/run.php
index 342a31f7..94705b04 100644
--- a/benchmark/run.php
+++ b/benchmark/run.php
@@ -28,7 +28,7 @@
             $files = array_merge($files, glob($fileOrDir . '/*.php'));
         } else {
             // Force absolute path
-            $files[] = $file[0] === DIRECTORY_SEPARATOR ? $file : $_SERVER['PWD'] . DIRECTORY_SEPARATOR . $file;
+            $files[] = $fileOrDir[0] === DIRECTORY_SEPARATOR ? $fileOrDir : $_SERVER['PWD'] . DIRECTORY_SEPARATOR . $fileOrDir;
         }
     }
 }
@@ -37,7 +37,7 @@
 Observable::just($files)
     ->doOnNext(function(array $files) {
         printf("Benchmarking %d file/s (min %ds each)\n", count($files), MIN_TOTAL_DURATION);
-        printf("script_name - total_runs (single_run_mean ±standard_deviation)\n");
+        printf("script_name - total_runs (single_run_mean ±standard_deviation) - mem_start [mem_100_iter] mem_end\n");
         printf("==============================================================\n");
     })
     ->concatMap(function($files) { // Flatten the array
@@ -63,6 +63,8 @@ function () { }
             throw new Exception("Unable to load file \"$file\"");
         }
 
+        $memoryUsage = [memory_get_usage()];
+
         while ($totalDuration < MIN_TOTAL_DURATION) {
             $start = microtime(true);
 
@@ -72,13 +74,20 @@ function () { }
 
             $durations[] = $duration * 1000;
             $totalDuration += $duration;
+
+            if (count($durations) === 100) {
+                $memoryUsage[] = memory_get_usage();
+            }
         }
 
+        $memoryUsage[] = memory_get_usage();
+
         ob_end_clean();
 
         return [
             'file' => $file,
             'durations' => $durations,
+            'memory_usage' => $memoryUsage,
         ];
     })
     ->doOnNext(function(array $result) { // Print the number of successful runs
@@ -94,13 +103,18 @@ function () { }
 
         return [
             'file' => $result['file'],
+            'memory_usage' => $result['memory_usage'],
             'mean' => $mean,
             'standard_deviation' => pow($variance / $count, 0.5),
         ];
     })
     ->subscribe(new CallbackObserver(
         function(array $result) {
-            printf(" (%.2fms ±%.2fms)\n", $result['mean'], $result['standard_deviation']);
+            printf(" (%.2fms ±%.2fms) - ", $result['mean'], $result['standard_deviation']);
+            foreach ($result['memory_usage'] as $memory) {
+                printf("%.2fMB ", $memory / pow(10, 6));
+            }
+            printf("\n");
         },
         function(\Exception $error) {
             printf("\nError: %s\n", $error->getMessage());

From 9a28ddcd11e3068df6ad9e2a2daa72f1616e00f5 Mon Sep 17 00:00:00 2001
From: Martin Sikora <martin.sikora.ahoj@gmail.com>
Date: Sat, 18 Mar 2017 14:07:12 +0100
Subject: [PATCH 10/15] refactored benchmark runner

---
 benchmark/bufferWithCount/bufferWithCount.php | 11 ++--
 .../bufferWithCount_eventloop.php             | 12 ++--
 benchmark/distinct/distinct.php               | 13 ++--
 benchmark/distinct/distinct_eventloop.php     | 16 ++---
 benchmark/filter/filter.php                   | 21 ++++---
 benchmark/filter/filter_eventloop.php         | 23 +++----
 benchmark/run.php                             | 62 ++++++++++++++-----
 benchmark/skipLast/skipLast.php               | 11 ++--
 benchmark/skipLast/skipLast_eventloop.php     | 13 ++--
 benchmark/takeLast/takeLast.php               |  9 +--
 benchmark/takeLast/takeLast_eventloop.php     | 13 ++--
 benchmark/zip/zip.php                         | 13 ++--
 benchmark/zip/zip_eventloop.php               | 17 ++---
 13 files changed, 137 insertions(+), 97 deletions(-)

diff --git a/benchmark/bufferWithCount/bufferWithCount.php b/benchmark/bufferWithCount/bufferWithCount.php
index acb4ca4b..937b10f6 100644
--- a/benchmark/bufferWithCount/bufferWithCount.php
+++ b/benchmark/bufferWithCount/bufferWithCount.php
@@ -2,8 +2,9 @@
 
 use Rx\Observable;
 
-return function() use ($dummyObserver) {
-    Observable::range(0, 25)
-        ->bufferWithCount(5)
-        ->subscribe($dummyObserver);
-};
\ No newline at end of file
+$source = Observable::range(0, 25)
+    ->bufferWithCount(5);
+
+return function() use ($source) {
+    return $source;
+};
diff --git a/benchmark/bufferWithCount/bufferWithCount_eventloop.php b/benchmark/bufferWithCount/bufferWithCount_eventloop.php
index 84cdcbd6..55fe8bea 100644
--- a/benchmark/bufferWithCount/bufferWithCount_eventloop.php
+++ b/benchmark/bufferWithCount/bufferWithCount_eventloop.php
@@ -6,11 +6,11 @@
 
 $loop = new StreamSelectLoop();
 $scheduler = new EventLoopScheduler($loop);
+$source = Observable::range(0, 25, $scheduler)
+    ->bufferWithCount(5);
 
-return function() use ($dummyObserver, $scheduler, $loop) {
-    Observable::range(0, 25)
-        ->bufferWithCount(5)
-        ->subscribe($dummyObserver, $scheduler);
+$factory = function() use ($source, $scheduler) {
+    return $source;
+};
 
-    $loop->run();
-};
\ No newline at end of file
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/distinct/distinct.php b/benchmark/distinct/distinct.php
index 120efca2..7c3ecd21 100644
--- a/benchmark/distinct/distinct.php
+++ b/benchmark/distinct/distinct.php
@@ -2,12 +2,13 @@
 
 use Rx\Observable;
 
-$source = array_map(function($val) {
+$range = array_map(function($val) {
     return $val % 3;
 }, range(0, 25));
 
-return function() use ($source, $dummyObserver) {
-    Observable::fromArray($source)
-        ->distinct()
-        ->subscribe($dummyObserver);
-};
\ No newline at end of file
+$source = Observable::fromArray($range)
+    ->distinct();
+
+return function() use ($source) {
+    return $source;
+};
diff --git a/benchmark/distinct/distinct_eventloop.php b/benchmark/distinct/distinct_eventloop.php
index 703e7c97..8e2e4d5e 100644
--- a/benchmark/distinct/distinct_eventloop.php
+++ b/benchmark/distinct/distinct_eventloop.php
@@ -1,20 +1,22 @@
 <?php
 
 use Rx\Observable;
+use Rx\Observable\ArrayObservable;
 use Rx\Scheduler\EventLoopScheduler;
 use React\EventLoop\StreamSelectLoop;
 
 $loop = new StreamSelectLoop();
 $scheduler = new EventLoopScheduler($loop);
 
-$source = array_map(function($val) {
+$range = array_map(function($val) {
     return $val % 3;
 }, range(0, 25));
 
-return function() use ($source, $dummyObserver, $scheduler, $loop) {
-    Observable::fromArray($source)
-        ->distinct()
-        ->subscribe($dummyObserver, $scheduler);
+$source = (new ArrayObservable($range, $scheduler))
+    ->distinct();
 
-    $loop->run();
-};
\ No newline at end of file
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/filter/filter.php b/benchmark/filter/filter.php
index cb352d06..e99b4643 100644
--- a/benchmark/filter/filter.php
+++ b/benchmark/filter/filter.php
@@ -2,13 +2,14 @@
 
 use Rx\Observable;
 
-return function() use ($dummyObserver) {
-    Observable::range(0, 50)
-        ->filter(function($value) {
-            return $value % 2 == 0;
-        })
-        ->filter(function($value) {
-            return $value % 10 == 0;
-        })
-        ->subscribe($dummyObserver);
-};
\ No newline at end of file
+$source = Observable::range(0, 50)
+    ->filter(function($value) {
+        return $value % 2 == 0;
+    })
+    ->filter(function($value) {
+        return $value % 10 == 0;
+    });
+
+return function() use ($source) {
+    return $source;
+};
diff --git a/benchmark/filter/filter_eventloop.php b/benchmark/filter/filter_eventloop.php
index 821d406e..a3ee08aa 100644
--- a/benchmark/filter/filter_eventloop.php
+++ b/benchmark/filter/filter_eventloop.php
@@ -7,15 +7,16 @@
 $loop = new StreamSelectLoop();
 $scheduler = new EventLoopScheduler($loop);
 
-return function() use ($dummyObserver, $scheduler, $loop) {
-    Observable::range(0, 50)
-        ->filter(function($value) {
-            return $value % 2 == 0;
-        })
-        ->filter(function($value) {
-            return $value % 10 == 0;
-        })
-        ->subscribe($dummyObserver, $scheduler);
+$source = Observable::range(0, 50, $scheduler)
+    ->filter(function($value) {
+        return $value % 2 == 0;
+    })
+    ->filter(function($value) {
+        return $value % 10 == 0;
+    });
 
-    $loop->run();
-};
\ No newline at end of file
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/run.php b/benchmark/run.php
index 94705b04..4f38b666 100644
--- a/benchmark/run.php
+++ b/benchmark/run.php
@@ -9,6 +9,7 @@
 
 use Rx\Observable;
 use Rx\Observer\CallbackObserver;
+use React\EventLoop\LoopInterface;
 
 // Check whether XDebug is enabled
 if (in_array('Xdebug', get_loaded_extensions(true))) {
@@ -47,37 +48,64 @@
         printf('%s', pathinfo($file, PATHINFO_FILENAME));
     })
     ->map(function($file) { // Run benchmark
-        $totalDuration = 0.0;
         $durations = [];
+        /** @var Observable $observable */
+        $observable = null;
+        /** @var LoopInterface $loop */
+        $loop = null;
+        /** @var callable(): Observable $sourceFactory */
+        $sourceFactory = null;
 
         ob_start();
 
-        $dummyObserver = new Rx\Observer\CallbackObserver(
-            function ($value) { },
-            function ($error) { },
-            function () { }
-        );
+        $testDef = @include $file;
 
-        $testClosure = @include $file;
-        if (!$testClosure) {
-            throw new Exception("Unable to load file \"$file\"");
+        if (is_array($testDef)) {
+            $sourceFactory = $testDef[0];
+            $loop = $testDef[1];
+        } elseif (is_callable($testDef)) {
+            $sourceFactory = $testDef;
+        } else {
+            throw new Exception("File \"$file\" doesn't contain a valid benchmark");
         }
 
         $memoryUsage = [memory_get_usage()];
 
-        while ($totalDuration < MIN_TOTAL_DURATION) {
-            $start = microtime(true);
-
-            $testClosure();
-
-            $duration = microtime(true) - $start;
+        $benchmarkLoop = function(Observable $observable) use (&$durations, &$memoryUsage) {
+            $dummyObserver = new Rx\Observer\CallbackObserver(
+                function ($value) { },
+                function ($error) { },
+                function () use (&$start, &$durations) {
+                    $durations[] = (microtime(true) - $start) * 1000;
+                }
+            );
 
-            $durations[] = $duration * 1000;
-            $totalDuration += $duration;
+            $start = microtime(true);
+            $observable->subscribe($dummyObserver);
 
             if (count($durations) === 100) {
                 $memoryUsage[] = memory_get_usage();
             }
+        };
+
+        $stopStartTime = microtime(true) + MIN_TOTAL_DURATION;
+
+        if ($loop) {
+            $reschedule = function() use (&$reschedule, $benchmarkLoop, $sourceFactory, $loop, $stopStartTime) {
+                $loop->futureTick(function () use (&$reschedule, $benchmarkLoop, $stopStartTime, $sourceFactory) {
+                    $benchmarkLoop($sourceFactory());
+                    if ($stopStartTime > microtime(true)) {
+                        $reschedule();
+                    }
+                });
+            };
+
+            $reschedule();
+            $loop->run();
+        } else {
+            while ($stopStartTime > microtime(true)) {
+                $benchmarkLoop($sourceFactory());
+            }
         }
 
         $memoryUsage[] = memory_get_usage();
diff --git a/benchmark/skipLast/skipLast.php b/benchmark/skipLast/skipLast.php
index 3250592f..d4ddf7d4 100644
--- a/benchmark/skipLast/skipLast.php
+++ b/benchmark/skipLast/skipLast.php
@@ -2,8 +2,9 @@
 
 use Rx\Observable;
 
-return function() use ($dummyObserver) {
-    Observable::range(0, 500)
-        ->skipLast(50)
-        ->subscribe($dummyObserver);
-};
\ No newline at end of file
+$source = Observable::range(0, 500)
+    ->skipLast(50);
+
+return function() use ($source) {
+    return $source;
+};
diff --git a/benchmark/skipLast/skipLast_eventloop.php b/benchmark/skipLast/skipLast_eventloop.php
index e1db6194..5916e2b4 100644
--- a/benchmark/skipLast/skipLast_eventloop.php
+++ b/benchmark/skipLast/skipLast_eventloop.php
@@ -7,10 +7,11 @@
 $loop = new StreamSelectLoop();
 $scheduler = new EventLoopScheduler($loop);
 
-return function() use ($dummyObserver, $scheduler, $loop) {
-    Observable::range(0, 500)
-        ->skipLast(50)
-        ->subscribe($dummyObserver, $scheduler);
+$source = Observable::range(0, 500, $scheduler)
+    ->skipLast(50);
 
-    $loop->run();
-};
\ No newline at end of file
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/takeLast/takeLast.php b/benchmark/takeLast/takeLast.php
index c7a2650e..703f3a5c 100644
--- a/benchmark/takeLast/takeLast.php
+++ b/benchmark/takeLast/takeLast.php
@@ -2,8 +2,9 @@
 
 use Rx\Observable;
 
-return function() use ($dummyObserver) {
-    Observable::range(0, 500)
-        ->takeLast(50)
-        ->subscribe($dummyObserver);
+$source = Observable::range(0, 500)
+    ->takeLast(50);
+
+return function() use ($source) {
+    return $source;
 };
\ No newline at end of file
diff --git a/benchmark/takeLast/takeLast_eventloop.php b/benchmark/takeLast/takeLast_eventloop.php
index 3954ea58..463ad8d2 100644
--- a/benchmark/takeLast/takeLast_eventloop.php
+++ b/benchmark/takeLast/takeLast_eventloop.php
@@ -7,10 +7,11 @@
 $loop = new StreamSelectLoop();
 $scheduler = new EventLoopScheduler($loop);
 
-return function() use ($dummyObserver, $scheduler, $loop) {
-    Observable::range(0, 500)
-        ->takeLast(50)
-        ->subscribe($dummyObserver, $scheduler);
+$source = Observable::range(0, 500, $scheduler)
+    ->takeLast(50);
 
-    $loop->run();
-};
\ No newline at end of file
+$factory = function() use ($source, $scheduler) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/zip/zip.php b/benchmark/zip/zip.php
index f8d11898..236cbbc3 100644
--- a/benchmark/zip/zip.php
+++ b/benchmark/zip/zip.php
@@ -2,10 +2,11 @@
 
 use Rx\Observable;
 
-return function() use ($dummyObserver) {
-    Observable::range(0, 25)
-        ->zip([Observable::range(0, 25)], function($a, $b) {
-            return $a + $b;
-        })
-        ->subscribe($dummyObserver);
+$source = Observable::range(0, 25)
+    ->zip([Observable::range(0, 25)], function ($a, $b) {
+        return $a + $b;
+    });
+
+return function() use ($source) {
+    return $source;
 };
\ No newline at end of file
diff --git a/benchmark/zip/zip_eventloop.php b/benchmark/zip/zip_eventloop.php
index 6010b631..b69ed2a1 100644
--- a/benchmark/zip/zip_eventloop.php
+++ b/benchmark/zip/zip_eventloop.php
@@ -7,12 +7,13 @@
 $loop = new StreamSelectLoop();
 $scheduler = new EventLoopScheduler($loop);
 
-return function() use ($dummyObserver, $scheduler, $loop) {
-    Observable::range(0, 25)
-        ->zip([Observable::range(0, 25)], function($a, $b) {
-            return $a + $b;
-        })
-        ->subscribe($dummyObserver, $scheduler);
+$source = Observable::range(0, 25, $scheduler)
+    ->zip([Observable::range(0, 25, $scheduler)], function($a, $b) {
+        return $a + $b;
+    });
 
-    $loop->run();
-};
\ No newline at end of file
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];

From cea5ee54c700502d2370235ec0a00a08fb653cf5 Mon Sep 17 00:00:00 2001
From: Martin Sikora <martin.sikora.ahoj@gmail.com>
Date: Tue, 21 Mar 2017 22:37:02 +0100
Subject: [PATCH 11/15] more benchmarks

---
 .../bufferWithCount_eventloop.php             |  5 +++--
 benchmark/concat/concat.php                   | 10 ++++++++++
 benchmark/concat/concat_eventloop.php         | 17 ++++++++++++++++
 benchmark/concatAll/concatAll.php             | 13 ++++++++++++
 benchmark/concatAll/concatAll_eventloop.php   | 20 +++++++++++++++++++
 benchmark/concatMap/concatMap.php             | 12 +++++++++++
 benchmark/concatMap/concatMap_eventloop.php   | 19 ++++++++++++++++++
 benchmark/delay/delay.php                     | 11 ++++++++++
 benchmark/delay/delay_eventloop.php           | 17 ++++++++++++++++
 benchmark/flatMap/flatMap.php                 | 12 +++++++++++
 benchmark/flatMap/flatMap_eventloop.php       | 19 ++++++++++++++++++
 benchmark/merge/merge.php                     | 10 ++++++++++
 benchmark/merge/merge_eventloop.php           | 17 ++++++++++++++++
 benchmark/mergeAll/mergeAll.php               | 13 ++++++++++++
 benchmark/mergeAll/mergeAll_eventloop.php     | 20 +++++++++++++++++++
 15 files changed, 213 insertions(+), 2 deletions(-)
 create mode 100644 benchmark/concat/concat.php
 create mode 100644 benchmark/concat/concat_eventloop.php
 create mode 100644 benchmark/concatAll/concatAll.php
 create mode 100644 benchmark/concatAll/concatAll_eventloop.php
 create mode 100644 benchmark/concatMap/concatMap.php
 create mode 100644 benchmark/concatMap/concatMap_eventloop.php
 create mode 100644 benchmark/delay/delay.php
 create mode 100644 benchmark/delay/delay_eventloop.php
 create mode 100644 benchmark/flatMap/flatMap.php
 create mode 100644 benchmark/flatMap/flatMap_eventloop.php
 create mode 100644 benchmark/merge/merge.php
 create mode 100644 benchmark/merge/merge_eventloop.php
 create mode 100644 benchmark/mergeAll/mergeAll.php
 create mode 100644 benchmark/mergeAll/mergeAll_eventloop.php

diff --git a/benchmark/bufferWithCount/bufferWithCount_eventloop.php b/benchmark/bufferWithCount/bufferWithCount_eventloop.php
index 55fe8bea..349bf0b1 100644
--- a/benchmark/bufferWithCount/bufferWithCount_eventloop.php
+++ b/benchmark/bufferWithCount/bufferWithCount_eventloop.php
@@ -6,8 +6,9 @@
 
 $loop = new StreamSelectLoop();
 $scheduler = new EventLoopScheduler($loop);
-$source = Observable::range(0, 25, $scheduler)
-    ->bufferWithCount(5);
+
+$source = Observable::just(25)
+    ->delay(0, $scheduler);
 
 $factory = function() use ($source, $scheduler) {
     return $source;
diff --git a/benchmark/concat/concat.php b/benchmark/concat/concat.php
new file mode 100644
index 00000000..dfa9c4d6
--- /dev/null
+++ b/benchmark/concat/concat.php
@@ -0,0 +1,10 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(1, 25)
+    ->concat(Observable::range(1, 25));
+
+return function() use ($source) {
+    return $source;
+};
diff --git a/benchmark/concat/concat_eventloop.php b/benchmark/concat/concat_eventloop.php
new file mode 100644
index 00000000..0ae9e6e4
--- /dev/null
+++ b/benchmark/concat/concat_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(1, 25, $scheduler)
+    ->concat(Observable::range(1, 25, $scheduler));
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/concatAll/concatAll.php b/benchmark/concatAll/concatAll.php
new file mode 100644
index 00000000..4d4c1fa8
--- /dev/null
+++ b/benchmark/concatAll/concatAll.php
@@ -0,0 +1,13 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(0, 25)
+    ->map(function() {
+        return Observable::range(0, 25);
+    })
+    ->concatAll();
+
+return function() use ($source) {
+    return $source;
+};
diff --git a/benchmark/concatAll/concatAll_eventloop.php b/benchmark/concatAll/concatAll_eventloop.php
new file mode 100644
index 00000000..ac23e661
--- /dev/null
+++ b/benchmark/concatAll/concatAll_eventloop.php
@@ -0,0 +1,20 @@
+<?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() use ($scheduler) {
+        return Observable::range(0, 25, $scheduler);
+    })
+    ->concatAll();
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/concatMap/concatMap.php b/benchmark/concatMap/concatMap.php
new file mode 100644
index 00000000..6a7451c7
--- /dev/null
+++ b/benchmark/concatMap/concatMap.php
@@ -0,0 +1,12 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(1, 25)
+    ->concatMap(function($x) {
+        return Observable::range($x, 25);
+    });
+
+return function() use ($source) {
+    return $source;
+};
diff --git a/benchmark/concatMap/concatMap_eventloop.php b/benchmark/concatMap/concatMap_eventloop.php
new file mode 100644
index 00000000..c0fdc987
--- /dev/null
+++ b/benchmark/concatMap/concatMap_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(1, 25, $scheduler)
+    ->concatMap(function($x) use ($scheduler) {
+        return Observable::range($x, 25, $scheduler);
+    });
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/delay/delay.php b/benchmark/delay/delay.php
new file mode 100644
index 00000000..c45dd4c1
--- /dev/null
+++ b/benchmark/delay/delay.php
@@ -0,0 +1,11 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$source = Observable::just(25)
+    ->delay(0, new ImmediateScheduler());
+
+return function() use ($source) {
+    return $source;
+};
diff --git a/benchmark/delay/delay_eventloop.php b/benchmark/delay/delay_eventloop.php
new file mode 100644
index 00000000..5e999d63
--- /dev/null
+++ b/benchmark/delay/delay_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::just(25)
+    ->delay(0, $scheduler);
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/flatMap/flatMap.php b/benchmark/flatMap/flatMap.php
new file mode 100644
index 00000000..54644671
--- /dev/null
+++ b/benchmark/flatMap/flatMap.php
@@ -0,0 +1,12 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(0, 25)
+    ->flatMap(function($x) {
+        return Observable::range($x, 25);
+    });
+
+return function() use ($source) {
+    return $source;
+};
diff --git a/benchmark/flatMap/flatMap_eventloop.php b/benchmark/flatMap/flatMap_eventloop.php
new file mode 100644
index 00000000..ca3e7880
--- /dev/null
+++ b/benchmark/flatMap/flatMap_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)
+    ->flatMap(function($x) use ($scheduler) {
+        return Observable::range($x, 25, $scheduler);
+    });
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/merge/merge.php b/benchmark/merge/merge.php
new file mode 100644
index 00000000..7336a24b
--- /dev/null
+++ b/benchmark/merge/merge.php
@@ -0,0 +1,10 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(1, 25)
+    ->merge(Observable::range(1, 25));
+
+return function() use ($source) {
+    return $source;
+};
diff --git a/benchmark/merge/merge_eventloop.php b/benchmark/merge/merge_eventloop.php
new file mode 100644
index 00000000..0e630330
--- /dev/null
+++ b/benchmark/merge/merge_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, 250, $scheduler)
+    ->merge(Observable::range(0, 250, $scheduler));
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/mergeAll/mergeAll.php b/benchmark/mergeAll/mergeAll.php
new file mode 100644
index 00000000..5c3ab329
--- /dev/null
+++ b/benchmark/mergeAll/mergeAll.php
@@ -0,0 +1,13 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(0, 25)
+    ->map(function() {
+        return Observable::range(0, 25);
+    })
+    ->mergeAll();
+
+return function() use ($source) {
+    return $source;
+};
diff --git a/benchmark/mergeAll/mergeAll_eventloop.php b/benchmark/mergeAll/mergeAll_eventloop.php
new file mode 100644
index 00000000..6e2fd002
--- /dev/null
+++ b/benchmark/mergeAll/mergeAll_eventloop.php
@@ -0,0 +1,20 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\EventLoopScheduler;
+use React\EventLoop\StreamSelectLoop;
+
+$loop = new StreamSelectLoop();
+$scheduler = new EventLoopScheduler($loop);
+
+$source = Observable::range(0, 25)
+    ->map(function() {
+        return Observable::range(0, 25);
+    })
+    ->mergeAll();
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file

From 02f79e445841cbf11088e335b5928fe756ba51d4 Mon Sep 17 00:00:00 2001
From: Martin Sikora <martin.sikora.ahoj@gmail.com>
Date: Wed, 22 Mar 2017 12:07:17 +0100
Subject: [PATCH 12/15] more benchmarks

---
 benchmark/catch/catch.php                     | 12 +++++++++
 benchmark/combineLatest/combineLatest.php     | 12 +++++++++
 .../combineLatest/combineLatest_eventloop.php | 19 ++++++++++++++
 benchmark/concat/concat.php                   |  4 +--
 benchmark/concat/concat_eventloop.php         |  4 +--
 benchmark/defaultIfEmpty/defaultIfEmpty.php   | 10 +++++++
 benchmark/distinct/distinct.php               |  9 +++----
 benchmark/distinct/distinct_eventloop.php     |  9 +++----
 benchmark/forkJoin/forkJoin.php               | 15 +++++++++++
 benchmark/groupBy/groupBy.php                 | 15 +++++++++++
 benchmark/groupBy/groupBy_eventloop.php       | 22 ++++++++++++++++
 benchmark/isEmpty/isEmpty.php                 | 10 +++++++
 benchmark/merge/merge.php                     |  4 +--
 benchmark/merge/merge_eventloop.php           |  4 +--
 benchmark/mergeAll/mergeAll_eventloop.php     |  6 ++---
 benchmark/pluck/pluck.php                     | 10 +++++++
 benchmark/reduce/reduce.php                   | 12 +++++++++
 benchmark/reduce/reduce_eventloop.php         | 19 ++++++++++++++
 benchmark/repeat/repeat.php                   | 10 +++++++
 benchmark/repeat/repeat_eventloop.php         | 17 ++++++++++++
 benchmark/retry/retry.php                     | 19 ++++++++++++++
 benchmark/retry/retry_eventloop.php           | 26 +++++++++++++++++++
 benchmark/scan/scan.php                       | 12 +++++++++
 benchmark/scan/scan_eventloop.php             | 19 ++++++++++++++
 benchmark/skip/skip.php                       | 10 +++++++
 benchmark/skip/skip_eventloop.php             | 17 ++++++++++++
 benchmark/skipUntil/skipUntil_eventloop.php   | 18 +++++++++++++
 benchmark/skipWhile/skipWhile.php             | 12 +++++++++
 benchmark/skipWhile/skipWhile_eventloop.php   | 19 ++++++++++++++
 benchmark/startWith/startWith.php             | 10 +++++++
 benchmark/startWith/startWith_aray.php        | 10 +++++++
 .../startWith/startWith_aray_eventloop.php    | 15 +++++++++++
 benchmark/startWith/startWith_eventloop.php   | 15 +++++++++++
 benchmark/switchMap/switchMap.php             | 12 +++++++++
 benchmark/switchMap/switchMap_eventloop.php   | 19 ++++++++++++++
 benchmark/take/take.php                       | 10 +++++++
 benchmark/take/take_eventloop.php             | 17 ++++++++++++
 benchmark/takeLast/takeLast_eventloop.php     |  2 +-
 benchmark/takeUntil/takeUntil_eventloop.php   | 18 +++++++++++++
 benchmark/takeWhile/takeWhile.php             | 12 +++++++++
 benchmark/takeWhile/takeWhile_eventloop.php   | 19 ++++++++++++++
 benchmark/timestamp/timestamp.php             | 10 +++++++
 benchmark/timestamp/timestamp_eventloop.php   | 17 ++++++++++++
 43 files changed, 539 insertions(+), 22 deletions(-)
 create mode 100644 benchmark/catch/catch.php
 create mode 100644 benchmark/combineLatest/combineLatest.php
 create mode 100644 benchmark/combineLatest/combineLatest_eventloop.php
 create mode 100644 benchmark/defaultIfEmpty/defaultIfEmpty.php
 create mode 100644 benchmark/forkJoin/forkJoin.php
 create mode 100644 benchmark/groupBy/groupBy.php
 create mode 100644 benchmark/groupBy/groupBy_eventloop.php
 create mode 100644 benchmark/isEmpty/isEmpty.php
 create mode 100644 benchmark/pluck/pluck.php
 create mode 100644 benchmark/reduce/reduce.php
 create mode 100644 benchmark/reduce/reduce_eventloop.php
 create mode 100644 benchmark/repeat/repeat.php
 create mode 100644 benchmark/repeat/repeat_eventloop.php
 create mode 100644 benchmark/retry/retry.php
 create mode 100644 benchmark/retry/retry_eventloop.php
 create mode 100644 benchmark/scan/scan.php
 create mode 100644 benchmark/scan/scan_eventloop.php
 create mode 100644 benchmark/skip/skip.php
 create mode 100644 benchmark/skip/skip_eventloop.php
 create mode 100644 benchmark/skipUntil/skipUntil_eventloop.php
 create mode 100644 benchmark/skipWhile/skipWhile.php
 create mode 100644 benchmark/skipWhile/skipWhile_eventloop.php
 create mode 100644 benchmark/startWith/startWith.php
 create mode 100644 benchmark/startWith/startWith_aray.php
 create mode 100644 benchmark/startWith/startWith_aray_eventloop.php
 create mode 100644 benchmark/startWith/startWith_eventloop.php
 create mode 100644 benchmark/switchMap/switchMap.php
 create mode 100644 benchmark/switchMap/switchMap_eventloop.php
 create mode 100644 benchmark/take/take.php
 create mode 100644 benchmark/take/take_eventloop.php
 create mode 100644 benchmark/takeUntil/takeUntil_eventloop.php
 create mode 100644 benchmark/takeWhile/takeWhile.php
 create mode 100644 benchmark/takeWhile/takeWhile_eventloop.php
 create mode 100644 benchmark/timestamp/timestamp.php
 create mode 100644 benchmark/timestamp/timestamp_eventloop.php

diff --git a/benchmark/catch/catch.php b/benchmark/catch/catch.php
new file mode 100644
index 00000000..26c2553b
--- /dev/null
+++ b/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;
+};
diff --git a/benchmark/combineLatest/combineLatest.php b/benchmark/combineLatest/combineLatest.php
new file mode 100644
index 00000000..bd49d574
--- /dev/null
+++ b/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;
+};
diff --git a/benchmark/combineLatest/combineLatest_eventloop.php b/benchmark/combineLatest/combineLatest_eventloop.php
new file mode 100644
index 00000000..941ca4a0
--- /dev/null
+++ b/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];
\ No newline at end of file
diff --git a/benchmark/concat/concat.php b/benchmark/concat/concat.php
index dfa9c4d6..07c0b7e7 100644
--- a/benchmark/concat/concat.php
+++ b/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;
diff --git a/benchmark/concat/concat_eventloop.php b/benchmark/concat/concat_eventloop.php
index 0ae9e6e4..26d2a0e5 100644
--- a/benchmark/concat/concat_eventloop.php
+++ b/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;
diff --git a/benchmark/defaultIfEmpty/defaultIfEmpty.php b/benchmark/defaultIfEmpty/defaultIfEmpty.php
new file mode 100644
index 00000000..94cf1ed8
--- /dev/null
+++ b/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;
+};
diff --git a/benchmark/distinct/distinct.php b/benchmark/distinct/distinct.php
index 7c3ecd21..2a172a22 100644
--- a/benchmark/distinct/distinct.php
+++ b/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) {
diff --git a/benchmark/distinct/distinct_eventloop.php b/benchmark/distinct/distinct_eventloop.php
index 8e2e4d5e..8e1dc61d 100644
--- a/benchmark/distinct/distinct_eventloop.php
+++ b/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) {
diff --git a/benchmark/forkJoin/forkJoin.php b/benchmark/forkJoin/forkJoin.php
new file mode 100644
index 00000000..e36412ce
--- /dev/null
+++ b/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;
+};
diff --git a/benchmark/groupBy/groupBy.php b/benchmark/groupBy/groupBy.php
new file mode 100644
index 00000000..232925cb
--- /dev/null
+++ b/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;
+};
\ No newline at end of file
diff --git a/benchmark/groupBy/groupBy_eventloop.php b/benchmark/groupBy/groupBy_eventloop.php
new file mode 100644
index 00000000..374845f1
--- /dev/null
+++ b/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];
\ No newline at end of file
diff --git a/benchmark/isEmpty/isEmpty.php b/benchmark/isEmpty/isEmpty.php
new file mode 100644
index 00000000..c4546474
--- /dev/null
+++ b/benchmark/isEmpty/isEmpty.php
@@ -0,0 +1,10 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::just(25)
+    ->isEmpty();
+
+return function() use ($source) {
+    return $source;
+};
\ No newline at end of file
diff --git a/benchmark/merge/merge.php b/benchmark/merge/merge.php
index 7336a24b..85d5c10a 100644
--- a/benchmark/merge/merge.php
+++ b/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;
diff --git a/benchmark/merge/merge_eventloop.php b/benchmark/merge/merge_eventloop.php
index 0e630330..019cb04c 100644
--- a/benchmark/merge/merge_eventloop.php
+++ b/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;
diff --git a/benchmark/mergeAll/mergeAll_eventloop.php b/benchmark/mergeAll/mergeAll_eventloop.php
index 6e2fd002..785bd5a1 100644
--- a/benchmark/mergeAll/mergeAll_eventloop.php
+++ b/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();
 
diff --git a/benchmark/pluck/pluck.php b/benchmark/pluck/pluck.php
new file mode 100644
index 00000000..684c6063
--- /dev/null
+++ b/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;
+};
diff --git a/benchmark/reduce/reduce.php b/benchmark/reduce/reduce.php
new file mode 100644
index 00000000..4250dd92
--- /dev/null
+++ b/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;
+};
diff --git a/benchmark/reduce/reduce_eventloop.php b/benchmark/reduce/reduce_eventloop.php
new file mode 100644
index 00000000..d1b27b4f
--- /dev/null
+++ b/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];
\ No newline at end of file
diff --git a/benchmark/repeat/repeat.php b/benchmark/repeat/repeat.php
new file mode 100644
index 00000000..c84529ae
--- /dev/null
+++ b/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;
+};
diff --git a/benchmark/repeat/repeat_eventloop.php b/benchmark/repeat/repeat_eventloop.php
new file mode 100644
index 00000000..bb3ba758
--- /dev/null
+++ b/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];
\ No newline at end of file
diff --git a/benchmark/retry/retry.php b/benchmark/retry/retry.php
new file mode 100644
index 00000000..8a5d7596
--- /dev/null
+++ b/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;
+};
diff --git a/benchmark/retry/retry_eventloop.php b/benchmark/retry/retry_eventloop.php
new file mode 100644
index 00000000..4a56b6ee
--- /dev/null
+++ b/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];
\ No newline at end of file
diff --git a/benchmark/scan/scan.php b/benchmark/scan/scan.php
new file mode 100644
index 00000000..74ee7918
--- /dev/null
+++ b/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;
+};
diff --git a/benchmark/scan/scan_eventloop.php b/benchmark/scan/scan_eventloop.php
new file mode 100644
index 00000000..9745aee5
--- /dev/null
+++ b/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];
\ No newline at end of file
diff --git a/benchmark/skip/skip.php b/benchmark/skip/skip.php
new file mode 100644
index 00000000..4f45fe2e
--- /dev/null
+++ b/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;
+};
\ No newline at end of file
diff --git a/benchmark/skip/skip_eventloop.php b/benchmark/skip/skip_eventloop.php
new file mode 100644
index 00000000..118b6456
--- /dev/null
+++ b/benchmark/skip/skip_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, 50, $scheduler)
+    ->skip(25);
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/skipUntil/skipUntil_eventloop.php b/benchmark/skipUntil/skipUntil_eventloop.php
new file mode 100644
index 00000000..a32c833e
--- /dev/null
+++ b/benchmark/skipUntil/skipUntil_eventloop.php
@@ -0,0 +1,18 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\EventLoopScheduler;
+use React\EventLoop\StreamSelectLoop;
+
+$loop = new StreamSelectLoop();
+$scheduler = new EventLoopScheduler($loop);
+
+$range = Observable::range(0, 25, $scheduler);
+$source = $range
+    ->skipUntil($range->take(3));
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/skipWhile/skipWhile.php b/benchmark/skipWhile/skipWhile.php
new file mode 100644
index 00000000..cc81128b
--- /dev/null
+++ b/benchmark/skipWhile/skipWhile.php
@@ -0,0 +1,12 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(0, 50)
+    ->skipWhile(function($value) {
+        return $value < 25;
+    });
+
+return function() use ($source) {
+    return $source;
+};
diff --git a/benchmark/skipWhile/skipWhile_eventloop.php b/benchmark/skipWhile/skipWhile_eventloop.php
new file mode 100644
index 00000000..69bf26d0
--- /dev/null
+++ b/benchmark/skipWhile/skipWhile_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, 50, $scheduler)
+    ->skipWhile(function($value) {
+        return $value < 25;
+    });
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/startWith/startWith.php b/benchmark/startWith/startWith.php
new file mode 100644
index 00000000..0e4d879c
--- /dev/null
+++ b/benchmark/startWith/startWith.php
@@ -0,0 +1,10 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(0, 25)
+    ->startWith(5);
+
+return function() use ($source) {
+    return $source;
+};
\ No newline at end of file
diff --git a/benchmark/startWith/startWith_aray.php b/benchmark/startWith/startWith_aray.php
new file mode 100644
index 00000000..9d8671fd
--- /dev/null
+++ b/benchmark/startWith/startWith_aray.php
@@ -0,0 +1,10 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(0, 25)
+    ->startWithArray([5, 5, 5]);
+
+return function() use ($source) {
+    return $source;
+};
\ No newline at end of file
diff --git a/benchmark/startWith/startWith_aray_eventloop.php b/benchmark/startWith/startWith_aray_eventloop.php
new file mode 100644
index 00000000..7f128d33
--- /dev/null
+++ b/benchmark/startWith/startWith_aray_eventloop.php
@@ -0,0 +1,15 @@
+<?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)
+    ->startWithArray([5, 5, 5]);
+
+return function() use ($source) {
+    return $source;
+};
\ No newline at end of file
diff --git a/benchmark/startWith/startWith_eventloop.php b/benchmark/startWith/startWith_eventloop.php
new file mode 100644
index 00000000..d6285af4
--- /dev/null
+++ b/benchmark/startWith/startWith_eventloop.php
@@ -0,0 +1,15 @@
+<?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)
+    ->startWith(5);
+
+return function() use ($source) {
+    return $source;
+};
\ No newline at end of file
diff --git a/benchmark/switchMap/switchMap.php b/benchmark/switchMap/switchMap.php
new file mode 100644
index 00000000..be15d4a9
--- /dev/null
+++ b/benchmark/switchMap/switchMap.php
@@ -0,0 +1,12 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(0, 25)
+    ->flatMapLatest(function($x) {
+        return Observable::range($x, 25);
+    });
+
+return function() use ($source) {
+    return $source;
+};
diff --git a/benchmark/switchMap/switchMap_eventloop.php b/benchmark/switchMap/switchMap_eventloop.php
new file mode 100644
index 00000000..3adf6b4d
--- /dev/null
+++ b/benchmark/switchMap/switchMap_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)
+    ->flatMapLatest(function($x) use ($scheduler) {
+        return Observable::range($x, 25, $scheduler);
+    });
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/take/take.php b/benchmark/take/take.php
new file mode 100644
index 00000000..5d2f30d2
--- /dev/null
+++ b/benchmark/take/take.php
@@ -0,0 +1,10 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(0, 50)
+    ->take(5);
+
+return function() use ($source) {
+    return $source;
+};
\ No newline at end of file
diff --git a/benchmark/take/take_eventloop.php b/benchmark/take/take_eventloop.php
new file mode 100644
index 00000000..20bd0fbc
--- /dev/null
+++ b/benchmark/take/take_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, 50, $scheduler)
+    ->take(5);
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/takeLast/takeLast_eventloop.php b/benchmark/takeLast/takeLast_eventloop.php
index 463ad8d2..01082b7a 100644
--- a/benchmark/takeLast/takeLast_eventloop.php
+++ b/benchmark/takeLast/takeLast_eventloop.php
@@ -10,7 +10,7 @@
 $source = Observable::range(0, 500, $scheduler)
     ->takeLast(50);
 
-$factory = function() use ($source, $scheduler) {
+$factory = function() use ($source) {
     return $source;
 };
 
diff --git a/benchmark/takeUntil/takeUntil_eventloop.php b/benchmark/takeUntil/takeUntil_eventloop.php
new file mode 100644
index 00000000..57f406fb
--- /dev/null
+++ b/benchmark/takeUntil/takeUntil_eventloop.php
@@ -0,0 +1,18 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\EventLoopScheduler;
+use React\EventLoop\StreamSelectLoop;
+
+$loop = new StreamSelectLoop();
+$scheduler = new EventLoopScheduler($loop);
+
+$range = Observable::range(0, 25, $scheduler);
+$source = $range
+    ->takeUntil($range->take(3));
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/takeWhile/takeWhile.php b/benchmark/takeWhile/takeWhile.php
new file mode 100644
index 00000000..0cf54c2d
--- /dev/null
+++ b/benchmark/takeWhile/takeWhile.php
@@ -0,0 +1,12 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(0, 50)
+    ->takeWhile(function($value) {
+        return $value < 48;
+    });
+
+return function() use ($source) {
+    return $source;
+};
diff --git a/benchmark/takeWhile/takeWhile_eventloop.php b/benchmark/takeWhile/takeWhile_eventloop.php
new file mode 100644
index 00000000..9c636ec8
--- /dev/null
+++ b/benchmark/takeWhile/takeWhile_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, 50, $scheduler)
+    ->takeWhile(function($value) {
+        return $value < 48;
+    });
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/timestamp/timestamp.php b/benchmark/timestamp/timestamp.php
new file mode 100644
index 00000000..578c98dc
--- /dev/null
+++ b/benchmark/timestamp/timestamp.php
@@ -0,0 +1,10 @@
+<?php
+
+use Rx\Observable;
+
+$source = Observable::range(0, 5)
+    ->timestamp();
+
+return function() use ($source) {
+    return $source;
+};
diff --git a/benchmark/timestamp/timestamp_eventloop.php b/benchmark/timestamp/timestamp_eventloop.php
new file mode 100644
index 00000000..00681e42
--- /dev/null
+++ b/benchmark/timestamp/timestamp_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, 5, $scheduler)
+    ->timestamp();
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file

From 6d1294fce6b1b3fe00f0b2b5dae7d942805644ff Mon Sep 17 00:00:00 2001
From: Martin Sikora <martin.sikora.ahoj@gmail.com>
Date: Tue, 1 Aug 2017 16:29:56 +0200
Subject: [PATCH 13/15] use do() instead of doOnNext()

---
 benchmark/run.php | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/benchmark/run.php b/benchmark/run.php
index 4f38b666..be60c302 100644
--- a/benchmark/run.php
+++ b/benchmark/run.php
@@ -36,7 +36,7 @@
 
 
 Observable::just($files)
-    ->doOnNext(function(array $files) {
+    ->do(function(array $files) {
         printf("Benchmarking %d file/s (min %ds each)\n", count($files), MIN_TOTAL_DURATION);
         printf("script_name - total_runs (single_run_mean ±standard_deviation) - mem_start [mem_100_iter] mem_end\n");
         printf("==============================================================\n");
@@ -44,7 +44,7 @@
     ->concatMap(function($files) { // Flatten the array
         return Observable::fromArray($files);
     })
-    ->doOnNext(function($file) {
+    ->do(function($file) {
         printf('%s', pathinfo($file, PATHINFO_FILENAME));
     })
     ->map(function($file) { // Run benchmark
@@ -118,7 +118,7 @@ function () use (&$start, &$durations) {
             'memory_usage' => $memoryUsage,
         ];
     })
-    ->doOnNext(function(array $result) { // Print the number of successful runs
+    ->do(function(array $result) { // Print the number of successful runs
         printf(' - %d', count($result['durations']));
     })
     ->map(function(array $result) { // Calculate the standard deviation

From b81a07340a79b01c9e4afe37fe6cc1419f4a830b Mon Sep 17 00:00:00 2001
From: Martin Sikora <martin.sikora.ahoj@gmail.com>
Date: Tue, 1 Aug 2017 19:09:26 +0200
Subject: [PATCH 14/15] update benchmark runner for 2.x

---
 benchmark/delay/delay.php | 11 -----------
 benchmark/run.php         | 25 ++++++++++++++-----------
 2 files changed, 14 insertions(+), 22 deletions(-)
 delete mode 100644 benchmark/delay/delay.php

diff --git a/benchmark/delay/delay.php b/benchmark/delay/delay.php
deleted file mode 100644
index c45dd4c1..00000000
--- a/benchmark/delay/delay.php
+++ /dev/null
@@ -1,11 +0,0 @@
-<?php
-
-use Rx\Observable;
-use Rx\Scheduler\ImmediateScheduler;
-
-$source = Observable::just(25)
-    ->delay(0, new ImmediateScheduler());
-
-return function() use ($source) {
-    return $source;
-};
diff --git a/benchmark/run.php b/benchmark/run.php
index be60c302..dc866a61 100644
--- a/benchmark/run.php
+++ b/benchmark/run.php
@@ -10,6 +10,10 @@
 use Rx\Observable;
 use Rx\Observer\CallbackObserver;
 use React\EventLoop\LoopInterface;
+use React\EventLoop\Factory;
+use Rx\Scheduler;
+
+$scheduler = new Scheduler\ImmediateScheduler();
 
 // Check whether XDebug is enabled
 if (in_array('Xdebug', get_loaded_extensions(true))) {
@@ -34,15 +38,14 @@
     }
 }
 
-
-Observable::just($files)
+Observable::of($files, $scheduler)
     ->do(function(array $files) {
         printf("Benchmarking %d file/s (min %ds each)\n", count($files), MIN_TOTAL_DURATION);
         printf("script_name - total_runs (single_run_mean ±standard_deviation) - mem_start [mem_100_iter] mem_end\n");
         printf("==============================================================\n");
     })
-    ->concatMap(function($files) { // Flatten the array
-        return Observable::fromArray($files);
+    ->concatMap(function($files) use ($scheduler) { // Flatten the array
+        return Observable::fromArray($files, $scheduler);
     })
     ->do(function($file) {
         printf('%s', pathinfo($file, PATHINFO_FILENAME));
@@ -88,13 +91,13 @@ function () use (&$start, &$durations) {
             }
         };
 
-        $stopStartTime = microtime(true) + MIN_TOTAL_DURATION;
+        $stopTime = microtime(true) + MIN_TOTAL_DURATION;
 
         if ($loop) {
-            $reschedule = function() use (&$reschedule, $benchmarkLoop, $sourceFactory, $loop, $stopStartTime) {
-                $loop->futureTick(function () use (&$reschedule, $benchmarkLoop, $stopStartTime, $sourceFactory) {
+            $reschedule = function() use (&$reschedule, $benchmarkLoop, $sourceFactory, $loop, $stopTime) {
+                $loop->futureTick(function () use (&$reschedule, $benchmarkLoop, $stopTime, $sourceFactory) {
                     $benchmarkLoop($sourceFactory());
-                    if ($stopStartTime > microtime(true)) {
+                    if ($stopTime > microtime(true)) {
                         $reschedule();
                     }
                 });
@@ -103,7 +106,7 @@ function () use (&$start, &$durations) {
             $reschedule();
             $loop->run();
         } else {
-            while ($stopStartTime > microtime(true)) {
+            while ($stopTime > microtime(true)) {
                 $benchmarkLoop($sourceFactory());
             }
         }
@@ -136,7 +139,7 @@ function () use (&$start, &$durations) {
             'standard_deviation' => pow($variance / $count, 0.5),
         ];
     })
-    ->subscribe(new CallbackObserver(
+    ->subscribe(
         function(array $result) {
             printf(" (%.2fms ±%.2fms) - ", $result['mean'], $result['standard_deviation']);
             foreach ($result['memory_usage'] as $memory) {
@@ -151,4 +154,4 @@ function() use ($start) {
             printf("============================================================\n");
             printf("total duration: %.2fs\n", microtime(true) - $start);
         }
-    ));
+    );

From 0adaa5afd461d90ea01d4f32a991253b6a7991bb Mon Sep 17 00:00:00 2001
From: Martin Sikora <martin.sikora.ahoj@gmail.com>
Date: Tue, 1 Aug 2017 19:09:57 +0200
Subject: [PATCH 15/15] add ImmediateSchedulers where necessary

---
 benchmark/bufferWithCount/bufferWithCount.php   |  3 ++-
 .../bufferWithCount_eventloop.php               |  6 +++---
 benchmark/catch/catch.php                       |  9 ++++++---
 benchmark/catch/catch_eventloop.php             | 17 +++++++++++++++++
 benchmark/combineLatest/combineLatest.php       |  7 +++++--
 benchmark/concat/concat.php                     |  7 +++++--
 benchmark/concatAll/concatAll.php               |  9 ++++++---
 benchmark/concatMap/concatMap.php               |  9 ++++++---
 benchmark/defaultIfEmpty/defaultIfEmpty.php     |  7 +++++--
 .../defaultIfEmpty/defaultIfEmpty_eventloop.php | 15 +++++++++++++++
 benchmark/delay/delay_eventloop.php             |  2 +-
 benchmark/distinct/distinct.php                 |  5 ++++-
 benchmark/filter/filter.php                     |  5 ++++-
 benchmark/flatMap/flatMap.php                   |  9 ++++++---
 benchmark/forkJoin/forkJoin.php                 | 13 ++++++++-----
 benchmark/fromArray/fromArray.php               | 12 ++++++++++++
 benchmark/fromArray/fromArray_eventloop.php     | 16 ++++++++++++++++
 benchmark/groupBy/groupBy.php                   |  5 ++++-
 benchmark/isEmpty/isEmpty.php                   |  5 ++++-
 benchmark/merge/merge.php                       |  7 +++++--
 benchmark/mergeAll/mergeAll.php                 |  9 ++++++---
 benchmark/pluck/pluck.php                       |  5 ++++-
 benchmark/pluck/pluck_eventloop.php             | 15 +++++++++++++++
 benchmark/reduce/reduce.php                     |  5 ++++-
 benchmark/repeat/repeat.php                     |  5 ++++-
 benchmark/retry/retry.php                       |  7 +++++--
 benchmark/retry/retry_eventloop.php             |  2 +-
 benchmark/scan/scan.php                         |  7 +++++--
 benchmark/scan/scan_eventloop.php               |  2 +-
 benchmark/skip/skip.php                         |  5 ++++-
 benchmark/skipLast/skipLast.php                 |  5 ++++-
 benchmark/skipUntil/skipUntil.php               | 16 ++++++++++++++++
 benchmark/skipWhile/skipWhile.php               |  5 ++++-
 benchmark/startWith/startWith.php               |  7 +++++--
 benchmark/startWith/startWith_aray.php          |  7 +++++--
 .../startWith/startWith_aray_eventloop.php      |  2 +-
 benchmark/startWith/startWith_eventloop.php     |  2 +-
 benchmark/switchMap/switchMap.php               |  9 ++++++---
 benchmark/take/take.php                         |  5 ++++-
 benchmark/takeLast/takeLast.php                 |  5 ++++-
 benchmark/takeUntil/takeUntil.php               | 16 ++++++++++++++++
 benchmark/takeWhile/takeWhile.php               |  5 ++++-
 benchmark/timestamp/timestamp.php               |  7 +++++--
 benchmark/timestamp/timestamp_eventloop.php     |  2 +-
 benchmark/zip/zip.php                           |  7 +++++--
 45 files changed, 264 insertions(+), 66 deletions(-)
 create mode 100644 benchmark/catch/catch_eventloop.php
 create mode 100644 benchmark/defaultIfEmpty/defaultIfEmpty_eventloop.php
 create mode 100644 benchmark/fromArray/fromArray.php
 create mode 100644 benchmark/fromArray/fromArray_eventloop.php
 create mode 100644 benchmark/pluck/pluck_eventloop.php
 create mode 100644 benchmark/skipUntil/skipUntil.php
 create mode 100644 benchmark/takeUntil/takeUntil.php

diff --git a/benchmark/bufferWithCount/bufferWithCount.php b/benchmark/bufferWithCount/bufferWithCount.php
index 937b10f6..a5422285 100644
--- a/benchmark/bufferWithCount/bufferWithCount.php
+++ b/benchmark/bufferWithCount/bufferWithCount.php
@@ -1,8 +1,9 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 25)
+$source = Observable::range(0, 25, new ImmediateScheduler())
     ->bufferWithCount(5);
 
 return function() use ($source) {
diff --git a/benchmark/bufferWithCount/bufferWithCount_eventloop.php b/benchmark/bufferWithCount/bufferWithCount_eventloop.php
index 349bf0b1..752fa556 100644
--- a/benchmark/bufferWithCount/bufferWithCount_eventloop.php
+++ b/benchmark/bufferWithCount/bufferWithCount_eventloop.php
@@ -7,10 +7,10 @@
 $loop = new StreamSelectLoop();
 $scheduler = new EventLoopScheduler($loop);
 
-$source = Observable::just(25)
-    ->delay(0, $scheduler);
+$source = Observable::range(0, 25, $scheduler)
+    ->bufferWithCount(5);
 
-$factory = function() use ($source, $scheduler) {
+$factory = function() use ($source) {
     return $source;
 };
 
diff --git a/benchmark/catch/catch.php b/benchmark/catch/catch.php
index 26c2553b..8ecf3a46 100644
--- a/benchmark/catch/catch.php
+++ b/benchmark/catch/catch.php
@@ -1,10 +1,13 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::error(new \Exception('error'))
-    ->catchError(function() {
-        return Observable::just(25);
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::error(new \Exception('error'), $scheduler)
+    ->catch(function() use ($scheduler) {
+        return Observable::of(25, $scheduler);
     });
 
 return function() use ($source) {
diff --git a/benchmark/catch/catch_eventloop.php b/benchmark/catch/catch_eventloop.php
new file mode 100644
index 00000000..275fb1b0
--- /dev/null
+++ b/benchmark/catch/catch_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::error(new \Exception('error'), $scheduler)
+    ->catch(function() use ($scheduler) {
+        return Observable::of(25, $scheduler);
+    });
+
+return function() use ($source) {
+    return $source;
+};
diff --git a/benchmark/combineLatest/combineLatest.php b/benchmark/combineLatest/combineLatest.php
index bd49d574..48ada3f8 100644
--- a/benchmark/combineLatest/combineLatest.php
+++ b/benchmark/combineLatest/combineLatest.php
@@ -1,9 +1,12 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 25)
-    ->combineLatest([Observable::range(0, 25)], function($a, $b) {
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
+    ->combineLatest([Observable::range(0, 25, $scheduler)], function($a, $b) {
         return $a + $b;
     });
 
diff --git a/benchmark/concat/concat.php b/benchmark/concat/concat.php
index 07c0b7e7..81c5fda9 100644
--- a/benchmark/concat/concat.php
+++ b/benchmark/concat/concat.php
@@ -1,9 +1,12 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 25)
-    ->concat(Observable::range(0, 25));
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
+    ->concat(Observable::range(0, 25, $scheduler));
 
 return function() use ($source) {
     return $source;
diff --git a/benchmark/concatAll/concatAll.php b/benchmark/concatAll/concatAll.php
index 4d4c1fa8..5e55e86f 100644
--- a/benchmark/concatAll/concatAll.php
+++ b/benchmark/concatAll/concatAll.php
@@ -1,10 +1,13 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 25)
-    ->map(function() {
-        return Observable::range(0, 25);
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
+    ->map(function() use ($scheduler) {
+        return Observable::range(0, 25, $scheduler);
     })
     ->concatAll();
 
diff --git a/benchmark/concatMap/concatMap.php b/benchmark/concatMap/concatMap.php
index 6a7451c7..e1a0b82c 100644
--- a/benchmark/concatMap/concatMap.php
+++ b/benchmark/concatMap/concatMap.php
@@ -1,10 +1,13 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(1, 25)
-    ->concatMap(function($x) {
-        return Observable::range($x, 25);
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(1, 25, $scheduler)
+    ->concatMap(function($x) use ($scheduler) {
+        return Observable::range($x, 25, $scheduler);
     });
 
 return function() use ($source) {
diff --git a/benchmark/defaultIfEmpty/defaultIfEmpty.php b/benchmark/defaultIfEmpty/defaultIfEmpty.php
index 94cf1ed8..483089f8 100644
--- a/benchmark/defaultIfEmpty/defaultIfEmpty.php
+++ b/benchmark/defaultIfEmpty/defaultIfEmpty.php
@@ -1,9 +1,12 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::emptyObservable()
-    ->defaultIfEmpty(Observable::just(25));
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::empty($scheduler)
+    ->defaultIfEmpty(Observable::of(25, $scheduler));
 
 return function() use ($source) {
     return $source;
diff --git a/benchmark/defaultIfEmpty/defaultIfEmpty_eventloop.php b/benchmark/defaultIfEmpty/defaultIfEmpty_eventloop.php
new file mode 100644
index 00000000..670fb105
--- /dev/null
+++ b/benchmark/defaultIfEmpty/defaultIfEmpty_eventloop.php
@@ -0,0 +1,15 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\EventLoopScheduler;
+use React\EventLoop\StreamSelectLoop;
+
+$loop = new StreamSelectLoop();
+$scheduler = new EventLoopScheduler($loop);
+
+$source = Observable::empty($scheduler)
+    ->defaultIfEmpty(Observable::of(25, $scheduler));
+
+return function() use ($source) {
+    return $source;
+};
diff --git a/benchmark/delay/delay_eventloop.php b/benchmark/delay/delay_eventloop.php
index 5e999d63..2f5431d1 100644
--- a/benchmark/delay/delay_eventloop.php
+++ b/benchmark/delay/delay_eventloop.php
@@ -7,7 +7,7 @@
 $loop = new StreamSelectLoop();
 $scheduler = new EventLoopScheduler($loop);
 
-$source = Observable::just(25)
+$source = Observable::of(25, $scheduler)
     ->delay(0, $scheduler);
 
 $factory = function() use ($source) {
diff --git a/benchmark/distinct/distinct.php b/benchmark/distinct/distinct.php
index 2a172a22..8117915b 100644
--- a/benchmark/distinct/distinct.php
+++ b/benchmark/distinct/distinct.php
@@ -1,8 +1,11 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 25)
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
     ->map(function($i) {
         return $i % 3;
     })
diff --git a/benchmark/filter/filter.php b/benchmark/filter/filter.php
index e99b4643..1e3fbb87 100644
--- a/benchmark/filter/filter.php
+++ b/benchmark/filter/filter.php
@@ -1,8 +1,11 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 50)
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 50, $scheduler)
     ->filter(function($value) {
         return $value % 2 == 0;
     })
diff --git a/benchmark/flatMap/flatMap.php b/benchmark/flatMap/flatMap.php
index 54644671..7f36ed78 100644
--- a/benchmark/flatMap/flatMap.php
+++ b/benchmark/flatMap/flatMap.php
@@ -1,10 +1,13 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 25)
-    ->flatMap(function($x) {
-        return Observable::range($x, 25);
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
+    ->flatMap(function($x) use ($scheduler) {
+        return Observable::range($x, 25, $scheduler);
     });
 
 return function() use ($source) {
diff --git a/benchmark/forkJoin/forkJoin.php b/benchmark/forkJoin/forkJoin.php
index e36412ce..ad5053db 100644
--- a/benchmark/forkJoin/forkJoin.php
+++ b/benchmark/forkJoin/forkJoin.php
@@ -1,14 +1,17 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::defer(function() {
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::defer(function() use ($scheduler) {
     return Observable::forkJoin([
-        Observable::just(25),
-        Observable::range(0, 25),
-        Observable::fromArray(([1, 2, 3, 4, 5]))
+        Observable::of(25, $scheduler),
+        Observable::range(0, 25, $scheduler),
+        Observable::fromArray([1, 2, 3, 4, 5], $scheduler)
     ]);
-});
+}, $scheduler);
 
 return function() use ($source) {
     return $source;
diff --git a/benchmark/fromArray/fromArray.php b/benchmark/fromArray/fromArray.php
new file mode 100644
index 00000000..7bcca0f7
--- /dev/null
+++ b/benchmark/fromArray/fromArray.php
@@ -0,0 +1,12 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::fromArray([0, 1, 2, 3, 4, 5, 6, 7, 8, 9], $scheduler);
+
+return [function() use ($source) {
+    return $source;
+}, $loop];
\ No newline at end of file
diff --git a/benchmark/fromArray/fromArray_eventloop.php b/benchmark/fromArray/fromArray_eventloop.php
new file mode 100644
index 00000000..1ed192c0
--- /dev/null
+++ b/benchmark/fromArray/fromArray_eventloop.php
@@ -0,0 +1,16 @@
+<?php
+
+use Rx\Observable;
+use React\EventLoop\StreamSelectLoop;
+use Rx\Scheduler;
+
+$loop = new StreamSelectLoop();
+$scheduler = new Scheduler\EventLoopScheduler($loop);
+
+$source = Observable::fromArray([0, 1, 2, 3, 4, 5, 6, 7, 8, 9], $scheduler);
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/groupBy/groupBy.php b/benchmark/groupBy/groupBy.php
index 232925cb..f6f5bb73 100644
--- a/benchmark/groupBy/groupBy.php
+++ b/benchmark/groupBy/groupBy.php
@@ -1,8 +1,11 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 25)
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
     ->map(function($i) {
         return ['key' => $i % 5];
     })
diff --git a/benchmark/isEmpty/isEmpty.php b/benchmark/isEmpty/isEmpty.php
index c4546474..e5af6732 100644
--- a/benchmark/isEmpty/isEmpty.php
+++ b/benchmark/isEmpty/isEmpty.php
@@ -1,8 +1,11 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::just(25)
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::of(25, $scheduler)
     ->isEmpty();
 
 return function() use ($source) {
diff --git a/benchmark/merge/merge.php b/benchmark/merge/merge.php
index 85d5c10a..52720c3d 100644
--- a/benchmark/merge/merge.php
+++ b/benchmark/merge/merge.php
@@ -1,9 +1,12 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 25)
-    ->merge(Observable::range(0, 25));
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
+    ->merge(Observable::range(0, 25, $scheduler));
 
 return function() use ($source) {
     return $source;
diff --git a/benchmark/mergeAll/mergeAll.php b/benchmark/mergeAll/mergeAll.php
index 5c3ab329..c5e12dc5 100644
--- a/benchmark/mergeAll/mergeAll.php
+++ b/benchmark/mergeAll/mergeAll.php
@@ -1,10 +1,13 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 25)
-    ->map(function() {
-        return Observable::range(0, 25);
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
+    ->map(function() use ($scheduler) {
+        return Observable::range(0, 25, $scheduler);
     })
     ->mergeAll();
 
diff --git a/benchmark/pluck/pluck.php b/benchmark/pluck/pluck.php
index 684c6063..6ba3dac2 100644
--- a/benchmark/pluck/pluck.php
+++ b/benchmark/pluck/pluck.php
@@ -1,8 +1,11 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::just([1, 2, 3, 4, 5])
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::of([1, 2, 3, 4, 5], $scheduler)
     ->pluck(2);
 
 return function() use ($source) {
diff --git a/benchmark/pluck/pluck_eventloop.php b/benchmark/pluck/pluck_eventloop.php
new file mode 100644
index 00000000..16a9d4c7
--- /dev/null
+++ b/benchmark/pluck/pluck_eventloop.php
@@ -0,0 +1,15 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\EventLoopScheduler;
+use React\EventLoop\StreamSelectLoop;
+
+$loop = new StreamSelectLoop();
+$scheduler = new EventLoopScheduler($loop);
+
+$source = Observable::of([1, 2, 3, 4, 5], $scheduler)
+    ->pluck(2);
+
+return function() use ($source) {
+    return $source;
+};
diff --git a/benchmark/reduce/reduce.php b/benchmark/reduce/reduce.php
index 4250dd92..a141733f 100644
--- a/benchmark/reduce/reduce.php
+++ b/benchmark/reduce/reduce.php
@@ -1,8 +1,11 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 25)
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
     ->reduce(function($a, $b) {
         return $a + $b;
     });
diff --git a/benchmark/repeat/repeat.php b/benchmark/repeat/repeat.php
index c84529ae..de8ad36e 100644
--- a/benchmark/repeat/repeat.php
+++ b/benchmark/repeat/repeat.php
@@ -1,8 +1,11 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 25)
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
     ->repeat(5);
 
 return function() use ($source) {
diff --git a/benchmark/retry/retry.php b/benchmark/retry/retry.php
index 8a5d7596..9b3a6d24 100644
--- a/benchmark/retry/retry.php
+++ b/benchmark/retry/retry.php
@@ -1,16 +1,19 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
 
 $maxRetryCount = 25;
 $newRetryCount = 0;
 
-$source = Observable::range(5, 1)
+$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);
+        return Observable::of($x);
     })
     ->retry($maxRetryCount);
 
diff --git a/benchmark/retry/retry_eventloop.php b/benchmark/retry/retry_eventloop.php
index 4a56b6ee..90ec0bda 100644
--- a/benchmark/retry/retry_eventloop.php
+++ b/benchmark/retry/retry_eventloop.php
@@ -15,7 +15,7 @@
         if (++$newRetryCount < $maxRetryCount - 1) {
             return Observable::error(new \Exception('error'));
         }
-        return Observable::just($x);
+        return Observable::of($x);
     })
     ->retry($maxRetryCount);
 
diff --git a/benchmark/scan/scan.php b/benchmark/scan/scan.php
index 74ee7918..432abc1c 100644
--- a/benchmark/scan/scan.php
+++ b/benchmark/scan/scan.php
@@ -1,10 +1,13 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 25)
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
     ->scan(function($acc, $x) {
-        return $x + $x;
+        return $acc + $x;
     });
 
 return function() use ($source) {
diff --git a/benchmark/scan/scan_eventloop.php b/benchmark/scan/scan_eventloop.php
index 9745aee5..b4814168 100644
--- a/benchmark/scan/scan_eventloop.php
+++ b/benchmark/scan/scan_eventloop.php
@@ -9,7 +9,7 @@
 
 $source = Observable::range(0, 25, $scheduler)
     ->scan(function($acc, $x) {
-        return $x + $x;
+        return $acc + $x;
     });
 
 $factory = function() use ($source) {
diff --git a/benchmark/skip/skip.php b/benchmark/skip/skip.php
index 4f45fe2e..8194417e 100644
--- a/benchmark/skip/skip.php
+++ b/benchmark/skip/skip.php
@@ -1,8 +1,11 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 50)
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 50, $scheduler)
     ->skip(25);
 
 return function() use ($source) {
diff --git a/benchmark/skipLast/skipLast.php b/benchmark/skipLast/skipLast.php
index d4ddf7d4..3859dff0 100644
--- a/benchmark/skipLast/skipLast.php
+++ b/benchmark/skipLast/skipLast.php
@@ -1,8 +1,11 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 500)
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 500, $scheduler)
     ->skipLast(50);
 
 return function() use ($source) {
diff --git a/benchmark/skipUntil/skipUntil.php b/benchmark/skipUntil/skipUntil.php
new file mode 100644
index 00000000..24dcd7b0
--- /dev/null
+++ b/benchmark/skipUntil/skipUntil.php
@@ -0,0 +1,16 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$range = Observable::range(0, 25, $scheduler);
+$source = $range
+    ->skipUntil($range->take(3));
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/skipWhile/skipWhile.php b/benchmark/skipWhile/skipWhile.php
index cc81128b..3d9af23c 100644
--- a/benchmark/skipWhile/skipWhile.php
+++ b/benchmark/skipWhile/skipWhile.php
@@ -1,8 +1,11 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 50)
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 50, $scheduler)
     ->skipWhile(function($value) {
         return $value < 25;
     });
diff --git a/benchmark/startWith/startWith.php b/benchmark/startWith/startWith.php
index 0e4d879c..3072ffc6 100644
--- a/benchmark/startWith/startWith.php
+++ b/benchmark/startWith/startWith.php
@@ -1,9 +1,12 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 25)
-    ->startWith(5);
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
+    ->startWith(5, $scheduler);
 
 return function() use ($source) {
     return $source;
diff --git a/benchmark/startWith/startWith_aray.php b/benchmark/startWith/startWith_aray.php
index 9d8671fd..f69dda50 100644
--- a/benchmark/startWith/startWith_aray.php
+++ b/benchmark/startWith/startWith_aray.php
@@ -1,9 +1,12 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 25)
-    ->startWithArray([5, 5, 5]);
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
+    ->startWithArray([5, 5, 5], $scheduler);
 
 return function() use ($source) {
     return $source;
diff --git a/benchmark/startWith/startWith_aray_eventloop.php b/benchmark/startWith/startWith_aray_eventloop.php
index 7f128d33..f33a50ff 100644
--- a/benchmark/startWith/startWith_aray_eventloop.php
+++ b/benchmark/startWith/startWith_aray_eventloop.php
@@ -8,7 +8,7 @@
 $scheduler = new EventLoopScheduler($loop);
 
 $source = Observable::range(0, 25, $scheduler)
-    ->startWithArray([5, 5, 5]);
+    ->startWithArray([5, 5, 5], $scheduler);
 
 return function() use ($source) {
     return $source;
diff --git a/benchmark/startWith/startWith_eventloop.php b/benchmark/startWith/startWith_eventloop.php
index d6285af4..059ae8ee 100644
--- a/benchmark/startWith/startWith_eventloop.php
+++ b/benchmark/startWith/startWith_eventloop.php
@@ -8,7 +8,7 @@
 $scheduler = new EventLoopScheduler($loop);
 
 $source = Observable::range(0, 25, $scheduler)
-    ->startWith(5);
+    ->startWith(5, $scheduler);
 
 return function() use ($source) {
     return $source;
diff --git a/benchmark/switchMap/switchMap.php b/benchmark/switchMap/switchMap.php
index be15d4a9..79272d13 100644
--- a/benchmark/switchMap/switchMap.php
+++ b/benchmark/switchMap/switchMap.php
@@ -1,10 +1,13 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 25)
-    ->flatMapLatest(function($x) {
-        return Observable::range($x, 25);
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
+    ->flatMapLatest(function($x) use ($scheduler) {
+        return Observable::range($x, 25, $scheduler);
     });
 
 return function() use ($source) {
diff --git a/benchmark/take/take.php b/benchmark/take/take.php
index 5d2f30d2..cc478f3f 100644
--- a/benchmark/take/take.php
+++ b/benchmark/take/take.php
@@ -1,8 +1,11 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 50)
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 50, $scheduler)
     ->take(5);
 
 return function() use ($source) {
diff --git a/benchmark/takeLast/takeLast.php b/benchmark/takeLast/takeLast.php
index 703f3a5c..7e906877 100644
--- a/benchmark/takeLast/takeLast.php
+++ b/benchmark/takeLast/takeLast.php
@@ -1,8 +1,11 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 500)
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 500, $scheduler)
     ->takeLast(50);
 
 return function() use ($source) {
diff --git a/benchmark/takeUntil/takeUntil.php b/benchmark/takeUntil/takeUntil.php
new file mode 100644
index 00000000..da46388b
--- /dev/null
+++ b/benchmark/takeUntil/takeUntil.php
@@ -0,0 +1,16 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$range = Observable::range(0, 25, $scheduler);
+$source = $range
+    ->takeUntil($range->take(3));
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/takeWhile/takeWhile.php b/benchmark/takeWhile/takeWhile.php
index 0cf54c2d..32106f9e 100644
--- a/benchmark/takeWhile/takeWhile.php
+++ b/benchmark/takeWhile/takeWhile.php
@@ -1,8 +1,11 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 50)
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 50, $scheduler)
     ->takeWhile(function($value) {
         return $value < 48;
     });
diff --git a/benchmark/timestamp/timestamp.php b/benchmark/timestamp/timestamp.php
index 578c98dc..5dd66a74 100644
--- a/benchmark/timestamp/timestamp.php
+++ b/benchmark/timestamp/timestamp.php
@@ -1,9 +1,12 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 5)
-    ->timestamp();
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 5, $scheduler)
+    ->timestamp($scheduler);
 
 return function() use ($source) {
     return $source;
diff --git a/benchmark/timestamp/timestamp_eventloop.php b/benchmark/timestamp/timestamp_eventloop.php
index 00681e42..ca7f6126 100644
--- a/benchmark/timestamp/timestamp_eventloop.php
+++ b/benchmark/timestamp/timestamp_eventloop.php
@@ -8,7 +8,7 @@
 $scheduler = new EventLoopScheduler($loop);
 
 $source = Observable::range(0, 5, $scheduler)
-    ->timestamp();
+    ->timestamp($scheduler);
 
 $factory = function() use ($source) {
     return $source;
diff --git a/benchmark/zip/zip.php b/benchmark/zip/zip.php
index 236cbbc3..4dfb34b7 100644
--- a/benchmark/zip/zip.php
+++ b/benchmark/zip/zip.php
@@ -1,9 +1,12 @@
 <?php
 
 use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
 
-$source = Observable::range(0, 25)
-    ->zip([Observable::range(0, 25)], function ($a, $b) {
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
+    ->zip([Observable::range(0, 25, $scheduler)], function ($a, $b) {
         return $a + $b;
     });