danog/amp
1
0
Fork 0
mirror of https://github.com/danog/amp.git synced 2024-11-27 04:24:42 +01:00
Code Issues Projects Releases Wiki Activity

Subscriber no longer implements Awaitable

Browse Source
This commit is contained in:
Aaron Piotrowski 2016-07-19 12:36:09 -05:00
parent 90d744537e
commit 32b4721c5e
4 changed files with 10 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
example/backpressure.php
View File

@ -15,12 +15,12 @@ Loop::execute(Amp\coroutine(function () {
$observable = $postponed->getObservable();
$subscriber = $observable->subscribe(function ($value) {
$observable->subscribe(function ($value) {
printf("Observable emitted %d\n", $value);
return new Pause(500); // Artificial back-pressure on observable.
});
$subscriber->when(function ($exception, $value) {
$observable->when(function ($exception, $value) {
if ($exception) {
printf("Observable failed: %s\n", $exception->getMessage());
return;

18
lib/Internal/Producer.php
View File

@ -67,13 +67,11 @@ trait Producer {
if ($this->result !== null) {
return new Subscriber(
$this->nextId++,
$this->result instanceof Awaitable ? $this->result : new Success($this->result),
$this->unsubscribe
);
}
$id = $this->nextId++;
$this->futures[$id] = $future = new Future;
$this->subscribers[$id] = $onNext;
if ($this->waiting !== null) {
@ -82,26 +80,23 @@ trait Producer {
$waiting->resolve();
}
return new Subscriber($id, $future, $this->unsubscribe);
return new Subscriber($id, $this->unsubscribe);
}
/**
* @param string $id
* @param \Throwable|\Exception|null $exception
*/
if (!isset($this->futures[$id])) {
private function unsubscribe($id, $exception = null) {
if (!isset($this->subscribers[$id])) {
return;
}
$future = $this->futures[$id];
unset($this->subscribers[$id], $this->futures[$id]);
unset($this->subscribers[$id]);
if (empty($this->subscribers)) {
$this->waiting = new Future;
}
$future->fail($exception ?: new UnsubscribedException);
}
/**
@ -196,9 +191,8 @@ trait Producer {
*
* @throws \LogicException If the observable has already been resolved.
*/
$futures = $this->futures;
$this->subscribers = $this->futures = [];
private function resolve($value = null) {
$this->subscribers = [];
if ($this->waiting !== null) {
$waiting = $this->waiting;
@ -207,9 +201,5 @@ trait Producer {
}
$this->complete($value);
foreach ($futures as $future) {
$future->resolve($value);
}
}
}

2
lib/Observer.php
View File

@ -78,7 +78,7 @@ class Observer {
$result = &$this->result;
$error = &$this->exception;
$this->subscriber->when(static function ($exception, $value) use (&$deferred, &$result, &$error, &$resolved) {
$observable->when(static function ($exception, $value) use (&$deferred, &$result, &$error, &$resolved) {
$resolved = true;
if ($exception) {

22
lib/Subscriber.php
View File

@ -2,22 +2,15 @@
namespace Amp;
use Interop\Async\Awaitable;
/**
* Subscriber implementation returned from implementors of \Amp\Observable.
*/
class Subscriber implements Awaitable {
class Subscriber {
/**
* @var string
*/
private $id;
/**
* @var \Interop\Async\Awaitable
*/
private $awaitable;
/**
* @var callable
*/
@ -25,24 +18,15 @@ class Subscriber implements Awaitable {
/**
* @param string $id
* @param \Interop\Async\Awaitable $awaitable
* @param callable $unsubscribe
*/
public function __construct($id, Awaitable $awaitable, callable $unsubscribe) {
public function __construct($id, callable $unsubscribe) {
$this->id = $id;
$this->awaitable = $awaitable;
$this->unsubscribe = $unsubscribe;
}
/**
* {@inheritdoc}
*/
public function when(callable $onResolved) {
$this->awaitable->when($onResolved);
}
/**
* {@inheritdoc}
* Unsubscribes from the Observable. No future values emitted by the Observable will be received.
*/
public function unsubscribe() {
$unsubscribe = $this->unsubscribe;