Rename Emitter to Producer; add new Emitter class

Emitter uses a coroutine to emit values. Updated examples.

example/emitter.php

@@ -3,34 +3,30 @@
 
 require dirname(__DIR__) . '/vendor/autoload.php';
 
+use Amp\Coroutine;
+use Amp\Emitter;
 use Amp\Observable;
 use Amp\Observer;
 use Amp\Pause;
-use Amp\Postponed;
 use Amp\Loop\NativeLoop;
 use Interop\Async\Loop;
 
 Loop::execute(Amp\coroutine(function () {
     try {
-        $postponed = new Postponed;
+        $emitter = new Emitter(function (callable $emit) {
-
+            yield $emit(1);
-        Loop::defer(function () use ($postponed) {
+            yield $emit(new Pause(500, 2));
-            // Observer emits all values at once.
+            yield $emit(3);
-            $postponed->emit(1);
+            yield $emit(new Pause(300, 4));
-            $postponed->emit(2);
+            yield $emit(5);
-            $postponed->emit(3);
+            yield $emit(6);
-            $postponed->emit(4);
+            yield $emit(new Pause(1000, 7));
-            $postponed->emit(5);
+            yield $emit(8);
-            $postponed->emit(6);
+            yield $emit(9);
-            $postponed->emit(7);
+            yield $emit(new Pause(600, 10));
-            $postponed->emit(8);
+            yield Coroutine::result(11);
-            $postponed->emit(9);
-            $postponed->emit(10);
-            $postponed->resolve(11);
         });
 
-        $observable = $postponed->getObservable();
-
         $generator = function (Observable $observable) {
             $observer = new Observer($observable);
 
@@ -42,7 +38,7 @@ Loop::execute(Amp\coroutine(function () {
             printf("Observable result %d\n", $observer->getResult());
         };
 
-        yield new \Amp\Coroutine($generator($observable));
+        yield new Coroutine($generator($emitter));
 
     } catch (\Exception $exception) {
         printf("Exception: %s\n", $exception);

example/postponed.php

@@ -0,0 +1,51 @@
+#!/usr/bin/env php
+<?php
+
+require dirname(__DIR__) . '/vendor/autoload.php';
+
+use Amp\Coroutine;
+use Amp\Observable;
+use Amp\Observer;
+use Amp\Pause;
+use Amp\Postponed;
+use Amp\Loop\NativeLoop;
+use Interop\Async\Loop;
+
+Loop::execute(Amp\coroutine(function () {
+    try {
+        $postponed = new Postponed;
+
+        Loop::defer(function () use ($postponed) {
+            // Observer emits all values at once.
+            $postponed->emit(1);
+            $postponed->emit(2);
+            $postponed->emit(3);
+            $postponed->emit(4);
+            $postponed->emit(5);
+            $postponed->emit(6);
+            $postponed->emit(7);
+            $postponed->emit(8);
+            $postponed->emit(9);
+            $postponed->emit(10);
+            $postponed->resolve(11);
+        });
+
+        $observable = $postponed->getObservable();
+
+        $generator = function (Observable $observable) {
+            $observer = new Observer($observable);
+
+            while (yield $observer->next()) {
+                printf("Observable emitted %d\n", $observer->getCurrent());
+                yield new Pause(100); // Observer consumption takes 100 ms.
+            }
+
+            printf("Observable result %d\n", $observer->getResult());
+        };
+
+        yield new Coroutine($generator($observable));
+
+    } catch (\Exception $exception) {
+        printf("Exception: %s\n", $exception);
+    }
+}), $loop = new NativeLoop());

lib/Emitter.php

@@ -2,14 +2,38 @@
 
 namespace Amp;
 
-/**
- * Observable implementation that should not be returned from a public API, but used only internally.
- */
 final class Emitter implements Observable {
-    use Internal\Producer {
+    use Internal\Producer;
-        init as __construct;
+
-        emit as public;
+    /**
-        resolve as public;
+     * @param callable(callable $emit): \Generator $emitter
-        fail as public;
+     */
+    public function __construct(callable $emitter) {
+        $this->init();
+    
+        /**
+         * @param mixed $value
+         *
+         * @return \Interop\Async\Awaitable
+         */
+        $emit = function ($value = null) {
+            return $this->emit($value);
+        };
+        
+        $result = $emitter($emit);
+
+        if (!$result instanceof \Generator) {
+            throw new \LogicException("The callable did not return a Generator");
+        }
+
+        $coroutine = new Coroutine($result);
+        $coroutine->when(function ($exception, $value) {
+            if ($exception) {
+                $this->fail($exception);
+                return;
+            }
+
+            $this->resolve($value);
+        });
     }
 }

lib/Producer.php

@@ -0,0 +1,15 @@
+<?php
+
+namespace Amp;
+
+/**
+ * Observable implementation that should not be returned from a public API, but used only internally.
+ */
+final class Producer implements Observable {
+    use Internal\Producer {
+        init as __construct;
+        emit as public;
+        resolve as public;
+        fail as public;
+    }
+}