async/src/functions.php at a37c28eeeb0611588e3ef5407fb105fed7f00434 · clue-labs/async · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
<?php

namespace React\Async;

/**
 * @param array<callable> $tasks
 * @param ?callable       $callback
 * @param ?callable       $errback
 * @return void
 */
function parallel(array $tasks, $callback = null, $errback = null)
{
    $results = array();
    $errors = array();

    $done = function () use (&$results, &$errors, $callback, $errback) {
        if (!$callback) {
            return;
        }

        if (count($errors)) {
            $errback(array_shift($errors));
            return;
        }

        $callback($results);
    };

    $numTasks = count($tasks);

    if (0 === $numTasks) {
        $done();
        return;
    }

    $checkDone = function () use (&$results, &$errors, $numTasks, $done) {
        if ($numTasks === count($results) + count($errors)) {
            $done();
        }
    };

    $taskErrback = function ($error) use (&$errors, $checkDone) {
        $errors[] = $error;
        $checkDone();
    };

    foreach ($tasks as $i => $task) {
        $taskCallback = function ($result) use (&$results, $i, $checkDone) {
            $results[$i] = $result;
            $checkDone();
        };

        call_user_func($task, $taskCallback, $taskErrback);
    }
}

/**
 * @param array<callable> $tasks
 * @param ?callable       $callback
 * @param ?callable       $errback
 * @return void
 */
function series(array $tasks, $callback = null, $errback = null)
{
    $results = array();

    /** @var callable():void $next */
    $taskCallback = function ($result) use (&$results, &$next) {
        $results[] = $result;
        $next();
    };

    $done = function () use (&$results, $callback) {
        if ($callback) {
            call_user_func($callback, $results);
        }
    };

    $next = function () use (&$tasks, $taskCallback, $errback, $done) {
        if (0 === count($tasks)) {
            $done();
            return;
        }

        $task = array_shift($tasks);
        call_user_func($task, $taskCallback, $errback);
    };

    $next();
}

/**
 * @param array<callable> $tasks
 * @param ?callable       $callback
 * @param ?callable       $errback
 * @return void
 */
function waterfall(array $tasks, $callback = null, $errback = null)
{
    $taskCallback = function () use (&$next) {
        call_user_func_array($next, func_get_args());
    };

    $done = function () use ($callback) {
        if ($callback) {
            call_user_func_array($callback, func_get_args());
        }
    };

    $next = function () use (&$tasks, $taskCallback, $errback, $done) {
        if (0 === count($tasks)) {
            call_user_func_array($done, func_get_args());
            return;
        }

        $task = array_shift($tasks);
        $args = array_merge(func_get_args(), array($taskCallback, $errback));
        call_user_func_array($task, $args);
    };

    $next();
}