diff --git a/benchmark/bufferWithCount/bufferWithCount.php b/benchmark/bufferWithCount/bufferWithCount.php
new file mode 100644
index 00000000..a5422285
--- /dev/null
+++ b/benchmark/bufferWithCount/bufferWithCount.php
@@ -0,0 +1,11 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$source = Observable::range(0, 25, new ImmediateScheduler())
+    ->bufferWithCount(5);
+
+return function() use ($source) {
+    return $source;
+};
diff --git a/benchmark/bufferWithCount/bufferWithCount_eventloop.php b/benchmark/bufferWithCount/bufferWithCount_eventloop.php
new file mode 100644
index 00000000..752fa556
--- /dev/null
+++ b/benchmark/bufferWithCount/bufferWithCount_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)
+    ->bufferWithCount(5);
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/catch/catch.php b/benchmark/catch/catch.php
new file mode 100644
index 00000000..8ecf3a46
--- /dev/null
+++ b/benchmark/catch/catch.php
@@ -0,0 +1,15 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$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/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
new file mode 100644
index 00000000..48ada3f8
--- /dev/null
+++ b/benchmark/combineLatest/combineLatest.php
@@ -0,0 +1,15 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
+    ->combineLatest([Observable::range(0, 25, $scheduler)], 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
new file mode 100644
index 00000000..81c5fda9
--- /dev/null
+++ b/benchmark/concat/concat.php
@@ -0,0 +1,13 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$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/concat/concat_eventloop.php b/benchmark/concat/concat_eventloop.php
new file mode 100644
index 00000000..26d2a0e5
--- /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(0, 25, $scheduler)
+    ->concat(Observable::range(0, 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..5e55e86f
--- /dev/null
+++ b/benchmark/concatAll/concatAll.php
@@ -0,0 +1,16 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
+    ->map(function() use ($scheduler) {
+        return Observable::range(0, 25, $scheduler);
+    })
+    ->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..e1a0b82c
--- /dev/null
+++ b/benchmark/concatMap/concatMap.php
@@ -0,0 +1,15 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(1, 25, $scheduler)
+    ->concatMap(function($x) use ($scheduler) {
+        return Observable::range($x, 25, $scheduler);
+    });
+
+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/defaultIfEmpty/defaultIfEmpty.php b/benchmark/defaultIfEmpty/defaultIfEmpty.php
new file mode 100644
index 00000000..483089f8
--- /dev/null
+++ b/benchmark/defaultIfEmpty/defaultIfEmpty.php
@@ -0,0 +1,13 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$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
new file mode 100644
index 00000000..2f5431d1
--- /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::of(25, $scheduler)
+    ->delay(0, $scheduler);
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/distinct/distinct.php b/benchmark/distinct/distinct.php
new file mode 100644
index 00000000..8117915b
--- /dev/null
+++ b/benchmark/distinct/distinct.php
@@ -0,0 +1,16 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
+    ->map(function($i) {
+        return $i % 3;
+    })
+    ->distinct();
+
+return function() use ($source) {
+    return $source;
+};
diff --git a/benchmark/distinct/distinct_eventloop.php b/benchmark/distinct/distinct_eventloop.php
new file mode 100644
index 00000000..8e1dc61d
--- /dev/null
+++ b/benchmark/distinct/distinct_eventloop.php
@@ -0,0 +1,21 @@
+<?php
+
+use Rx\Observable;
+use Rx\Observable\ArrayObservable;
+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 $i % 3;
+    })
+    ->distinct();
+
+$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
new file mode 100644
index 00000000..1e3fbb87
--- /dev/null
+++ b/benchmark/filter/filter.php
@@ -0,0 +1,18 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 50, $scheduler)
+    ->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
new file mode 100644
index 00000000..a3ee08aa
--- /dev/null
+++ b/benchmark/filter/filter_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, 50, $scheduler)
+    ->filter(function($value) {
+        return $value % 2 == 0;
+    })
+    ->filter(function($value) {
+        return $value % 10 == 0;
+    });
+
+$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..7f36ed78
--- /dev/null
+++ b/benchmark/flatMap/flatMap.php
@@ -0,0 +1,15 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
+    ->flatMap(function($x) use ($scheduler) {
+        return Observable::range($x, 25, $scheduler);
+    });
+
+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/forkJoin/forkJoin.php b/benchmark/forkJoin/forkJoin.php
new file mode 100644
index 00000000..ad5053db
--- /dev/null
+++ b/benchmark/forkJoin/forkJoin.php
@@ -0,0 +1,18 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::defer(function() use ($scheduler) {
+    return Observable::forkJoin([
+        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
new file mode 100644
index 00000000..f6f5bb73
--- /dev/null
+++ b/benchmark/groupBy/groupBy.php
@@ -0,0 +1,18 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
+    ->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..e5af6732
--- /dev/null
+++ b/benchmark/isEmpty/isEmpty.php
@@ -0,0 +1,13 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::of(25, $scheduler)
+    ->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
new file mode 100644
index 00000000..52720c3d
--- /dev/null
+++ b/benchmark/merge/merge.php
@@ -0,0 +1,13 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$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/merge/merge_eventloop.php b/benchmark/merge/merge_eventloop.php
new file mode 100644
index 00000000..019cb04c
--- /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, 25, $scheduler)
+    ->merge(Observable::range(0, 25, $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..c5e12dc5
--- /dev/null
+++ b/benchmark/mergeAll/mergeAll.php
@@ -0,0 +1,16 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
+    ->map(function() use ($scheduler) {
+        return Observable::range(0, 25, $scheduler);
+    })
+    ->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..785bd5a1
--- /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, $scheduler)
+    ->map(function() use ($scheduler) {
+        return Observable::range(0, 25, $scheduler);
+    })
+    ->mergeAll();
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/pluck/pluck.php b/benchmark/pluck/pluck.php
new file mode 100644
index 00000000..6ba3dac2
--- /dev/null
+++ b/benchmark/pluck/pluck.php
@@ -0,0 +1,13 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::of([1, 2, 3, 4, 5], $scheduler)
+    ->pluck(2);
+
+return function() use ($source) {
+    return $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
new file mode 100644
index 00000000..a141733f
--- /dev/null
+++ b/benchmark/reduce/reduce.php
@@ -0,0 +1,15 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
+    ->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..de8ad36e
--- /dev/null
+++ b/benchmark/repeat/repeat.php
@@ -0,0 +1,13 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
+    ->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..9b3a6d24
--- /dev/null
+++ b/benchmark/retry/retry.php
@@ -0,0 +1,22 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$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::of($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..90ec0bda
--- /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::of($x);
+    })
+    ->retry($maxRetryCount);
+
+$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
new file mode 100644
index 00000000..dc866a61
--- /dev/null
+++ b/benchmark/run.php
@@ -0,0 +1,157 @@
+<?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;
+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))) {
+    printf("Please, disable Xdebug extension before running RxPHP benchmarks.\n");
+    exit(1);
+}
+
+define('MIN_TOTAL_DURATION', 5);
+$start = microtime(true);
+
+if ($_SERVER['argc'] === 1) {
+    $files = glob(__DIR__ . '/**/*.php');
+} else {
+    $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[] = $fileOrDir[0] === DIRECTORY_SEPARATOR ? $fileOrDir : $_SERVER['PWD'] . DIRECTORY_SEPARATOR . $fileOrDir;
+        }
+    }
+}
+
+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) use ($scheduler) { // Flatten the array
+        return Observable::fromArray($files, $scheduler);
+    })
+    ->do(function($file) {
+        printf('%s', pathinfo($file, PATHINFO_FILENAME));
+    })
+    ->map(function($file) { // Run benchmark
+        $durations = [];
+        /** @var Observable $observable */
+        $observable = null;
+        /** @var LoopInterface $loop */
+        $loop = null;
+        /** @var callable(): Observable $sourceFactory */
+        $sourceFactory = null;
+
+        ob_start();
+
+        $testDef = @include $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()];
+
+        $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;
+                }
+            );
+
+            $start = microtime(true);
+            $observable->subscribe($dummyObserver);
+
+            if (count($durations) === 100) {
+                $memoryUsage[] = memory_get_usage();
+            }
+        };
+
+        $stopTime = microtime(true) + MIN_TOTAL_DURATION;
+
+        if ($loop) {
+            $reschedule = function() use (&$reschedule, $benchmarkLoop, $sourceFactory, $loop, $stopTime) {
+                $loop->futureTick(function () use (&$reschedule, $benchmarkLoop, $stopTime, $sourceFactory) {
+                    $benchmarkLoop($sourceFactory());
+                    if ($stopTime > microtime(true)) {
+                        $reschedule();
+                    }
+                });
+            };
+
+            $reschedule();
+            $loop->run();
+        } else {
+            while ($stopTime > microtime(true)) {
+                $benchmarkLoop($sourceFactory());
+            }
+        }
+
+        $memoryUsage[] = memory_get_usage();
+
+        ob_end_clean();
+
+        return [
+            'file' => $file,
+            'durations' => $durations,
+            'memory_usage' => $memoryUsage,
+        ];
+    })
+    ->do(function(array $result) { // Print the number of successful runs
+        printf(' - %d', count($result['durations']));
+    })
+    ->map(function(array $result) { // Calculate the standard deviation
+        $count = count($result['durations']);
+        $mean = array_sum($result['durations']) / $count;
+
+        $variance = array_sum(array_map(function($duration) use ($mean) {
+            return pow($mean - $duration, 2);
+        }, $result['durations']));
+
+        return [
+            'file' => $result['file'],
+            'memory_usage' => $result['memory_usage'],
+            'mean' => $mean,
+            'standard_deviation' => pow($variance / $count, 0.5),
+        ];
+    })
+    ->subscribe(
+        function(array $result) {
+            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());
+        },
+        function() use ($start) {
+            printf("============================================================\n");
+            printf("total duration: %.2fs\n", microtime(true) - $start);
+        }
+    );
diff --git a/benchmark/scan/scan.php b/benchmark/scan/scan.php
new file mode 100644
index 00000000..432abc1c
--- /dev/null
+++ b/benchmark/scan/scan.php
@@ -0,0 +1,15 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
+    ->scan(function($acc, $x) {
+        return $acc + $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..b4814168
--- /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 $acc + $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..8194417e
--- /dev/null
+++ b/benchmark/skip/skip.php
@@ -0,0 +1,13 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 50, $scheduler)
+    ->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/skipLast/skipLast.php b/benchmark/skipLast/skipLast.php
new file mode 100644
index 00000000..3859dff0
--- /dev/null
+++ b/benchmark/skipLast/skipLast.php
@@ -0,0 +1,13 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 500, $scheduler)
+    ->skipLast(50);
+
+return function() use ($source) {
+    return $source;
+};
diff --git a/benchmark/skipLast/skipLast_eventloop.php b/benchmark/skipLast/skipLast_eventloop.php
new file mode 100644
index 00000000..5916e2b4
--- /dev/null
+++ b/benchmark/skipLast/skipLast_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, 500, $scheduler)
+    ->skipLast(50);
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
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/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..3d9af23c
--- /dev/null
+++ b/benchmark/skipWhile/skipWhile.php
@@ -0,0 +1,15 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 50, $scheduler)
+    ->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..3072ffc6
--- /dev/null
+++ b/benchmark/startWith/startWith.php
@@ -0,0 +1,13 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
+    ->startWith(5, $scheduler);
+
+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..f69dda50
--- /dev/null
+++ b/benchmark/startWith/startWith_aray.php
@@ -0,0 +1,13 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
+    ->startWithArray([5, 5, 5], $scheduler);
+
+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..f33a50ff
--- /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], $scheduler);
+
+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..059ae8ee
--- /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, $scheduler);
+
+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..79272d13
--- /dev/null
+++ b/benchmark/switchMap/switchMap.php
@@ -0,0 +1,15 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
+    ->flatMapLatest(function($x) use ($scheduler) {
+        return Observable::range($x, 25, $scheduler);
+    });
+
+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..cc478f3f
--- /dev/null
+++ b/benchmark/take/take.php
@@ -0,0 +1,13 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 50, $scheduler)
+    ->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.php b/benchmark/takeLast/takeLast.php
new file mode 100644
index 00000000..7e906877
--- /dev/null
+++ b/benchmark/takeLast/takeLast.php
@@ -0,0 +1,13 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 500, $scheduler)
+    ->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
new file mode 100644
index 00000000..01082b7a
--- /dev/null
+++ b/benchmark/takeLast/takeLast_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, 500, $scheduler)
+    ->takeLast(50);
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
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/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..32106f9e
--- /dev/null
+++ b/benchmark/takeWhile/takeWhile.php
@@ -0,0 +1,15 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 50, $scheduler)
+    ->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..5dd66a74
--- /dev/null
+++ b/benchmark/timestamp/timestamp.php
@@ -0,0 +1,13 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$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
new file mode 100644
index 00000000..ca7f6126
--- /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($scheduler);
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];
\ No newline at end of file
diff --git a/benchmark/zip/zip.php b/benchmark/zip/zip.php
new file mode 100644
index 00000000..4dfb34b7
--- /dev/null
+++ b/benchmark/zip/zip.php
@@ -0,0 +1,15 @@
+<?php
+
+use Rx\Observable;
+use Rx\Scheduler\ImmediateScheduler;
+
+$scheduler = new ImmediateScheduler();
+
+$source = Observable::range(0, 25, $scheduler)
+    ->zip([Observable::range(0, 25, $scheduler)], 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
new file mode 100644
index 00000000..b69ed2a1
--- /dev/null
+++ b/benchmark/zip/zip_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)
+    ->zip([Observable::range(0, 25, $scheduler)], function($a, $b) {
+        return $a + $b;
+    });
+
+$factory = function() use ($source) {
+    return $source;
+};
+
+return [$factory, $loop];