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

Refactor functions with Emitter

Browse Source
This commit is contained in:
Aaron Piotrowski 2016-06-02 10:43:46 -05:00
parent cef5c90168
commit 5715bec9e1
1 changed files with 15 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
lib/functions.php
View File

@ -20,24 +20,23 @@ function merge(array $observables) {
}
}
$postponed = new Postponed;
return new Emitter(function (callable $emit) use ($observables) {
$subscriptions = [];
$subscriptions = [];
foreach ($observables as $observable) {
$subscriptions[] = $observable->subscribe([$postponed, 'emit']);
}
all($subscriptions)->when(function ($exception, $value) use ($postponed) {
if ($exception) {
$postponed->fail($exception);
return;
foreach ($observables as $observable) {
$subscriptions[] = $observable->subscribe($emit);
}
$postponed->resolve($value);
});
try {
$result = (yield all($subscriptions));
} finally {
foreach ($subscriptions as $subscription) {
$subscription->unsubscribe();
}
}
return $postponed->getObservable();
yield Coroutine::result($result);
});
}
/**
@ -90,23 +89,9 @@ function range($start, $end, $step = 1) {
throw new \InvalidArgumentException("Step is not of the correct sign");
}
$postponed = new Postponed;
$generator = function (Postponed $postponed, $start, $end, $step) {
return new Emitter(function (callable $emit) use ($start, $end, $step) {
for ($i = $start; $i <= $end; $i += $step) {
yield $postponed->emit($i);
yield $emit($i);
}
};
$coroutine = new Coroutine($generator($postponed, $start, $end, $step));
$coroutine->when(function ($exception) use ($postponed) {
if ($exception) {
$postponed->fail($exception);
return;
}
$postponed->resolve();
});
return $postponed->getObservable();
}