1
|
|
|
<?php |
2
|
|
|
|
3
|
|
|
namespace Dazzle\Filesystem\Invoker; |
4
|
|
|
|
5
|
|
|
use Dazzle\Filesystem\Driver\DriverAbstract; |
6
|
|
|
use Dazzle\Filesystem\Invoker\Queue\QueueCall; |
7
|
|
|
use Dazzle\Loop\LoopInterface; |
8
|
|
|
use Dazzle\Promise\Promise; |
9
|
|
|
use SplQueue; |
10
|
|
|
|
11
|
|
|
class InvokerQueue implements InvokerInterface |
12
|
|
|
{ |
13
|
|
|
/** |
14
|
|
|
* @var LoopInterface |
15
|
|
|
*/ |
16
|
|
|
protected $loop; |
17
|
|
|
|
18
|
|
|
/** |
19
|
|
|
* @var DriverAbstract |
20
|
|
|
*/ |
21
|
|
|
protected $driver; |
22
|
|
|
|
23
|
|
|
/** |
24
|
|
|
* @var SplQueue |
25
|
|
|
*/ |
26
|
|
|
protected $queue; |
27
|
|
|
|
28
|
|
|
/** |
29
|
|
|
* @param DriverAbstract $driver |
30
|
|
|
*/ |
31
|
|
|
public function __construct(DriverAbstract $driver) |
32
|
|
|
{ |
33
|
|
|
$this->loop = $driver->getLoop(); |
34
|
|
|
$this->driver = $driver; |
35
|
|
|
$this->queue = new SplQueue(); |
36
|
|
|
} |
37
|
|
|
|
38
|
|
|
/** |
39
|
|
|
* @override |
40
|
|
|
* @inheritDoc |
41
|
|
|
*/ |
42
|
|
|
public function call($func, $args = []) |
43
|
|
|
{ |
44
|
|
|
$promise = new Promise(); |
45
|
|
|
|
46
|
|
|
$this->queue->enqueue(new QueueCall($func, $args, $promise)); |
47
|
|
|
|
48
|
|
|
if (!$this->queue->isEmpty()) |
49
|
|
|
{ |
50
|
|
|
$this->processQueue(); |
51
|
|
|
} |
52
|
|
|
|
53
|
|
|
return $promise |
54
|
|
|
->then(function(QueueCall $call) { |
55
|
|
|
return $this->driver->call($call->func, $call->args); |
56
|
|
|
}) |
57
|
|
|
->then( |
58
|
|
|
$this->getQueueHandler('Dazzle\Promise\Promise::doResolve'), |
59
|
|
|
$this->getQueueHandler('Dazzle\Promise\Promise::doReject') |
60
|
|
|
); |
61
|
|
|
} |
62
|
|
|
|
63
|
|
|
/** |
64
|
|
|
* @override |
65
|
|
|
* @inheritDoc |
66
|
|
|
*/ |
67
|
|
|
public function isEmpty() |
68
|
|
|
{ |
69
|
|
|
return $this->queue->isEmpty(); |
70
|
|
|
} |
71
|
|
|
|
72
|
|
|
/** |
73
|
|
|
* Process queue. |
74
|
|
|
*/ |
75
|
|
|
protected function processQueue() |
76
|
|
|
{ |
77
|
|
|
$this->loop->onTick(function() { |
78
|
|
|
if ($this->queue->isEmpty()) |
79
|
|
|
{ |
80
|
|
|
return; |
81
|
|
|
} |
82
|
|
|
|
83
|
|
|
$call = $this->queue->dequeue(); |
84
|
|
|
$call->promise->resolve($call); |
85
|
|
|
}); |
86
|
|
|
} |
87
|
|
|
|
88
|
|
|
/** |
89
|
|
|
* Get queue handler. |
90
|
|
|
* |
91
|
|
|
* @param callable $func |
92
|
|
|
* @return callable |
93
|
|
|
*/ |
94
|
|
|
protected function getQueueHandler(callable $func) |
95
|
|
|
{ |
96
|
|
|
return function($mixed) use($func) { |
97
|
|
|
if (!$this->queue->isEmpty()) |
98
|
|
|
{ |
99
|
|
|
$this->processQueue(); |
100
|
|
|
} |
101
|
|
|
return $func($mixed); |
102
|
|
|
}; |
103
|
|
|
} |
104
|
|
|
} |
105
|
|
|
|