|
1
|
|
|
<?php |
|
2
|
|
|
|
|
3
|
|
|
namespace mpyw\Co\Internal; |
|
4
|
|
|
use mpyw\Co\CURLException; |
|
5
|
|
|
use mpyw\RuntimePromise\Deferred; |
|
6
|
|
|
|
|
7
|
|
|
class ManualScheduler extends AbstractScheduler |
|
8
|
|
|
{ |
|
9
|
|
|
/** |
|
10
|
|
|
* cURL handles those have not been dispatched. |
|
11
|
|
|
* @var array |
|
12
|
|
|
*/ |
|
13
|
|
|
private $queue = []; |
|
14
|
|
|
|
|
15
|
|
|
/** |
|
16
|
|
|
* TCP connection counter. |
|
17
|
|
|
* @var ConnectionCounter |
|
18
|
|
|
*/ |
|
19
|
|
|
private $counter; |
|
20
|
|
|
|
|
21
|
|
|
/** |
|
22
|
|
|
* Constructor. |
|
23
|
|
|
* Initialize cURL multi handle. |
|
24
|
|
|
* @param CoOption $options |
|
25
|
|
|
* @param resource $mh curl_multi |
|
26
|
|
|
*/ |
|
27
|
30 |
|
public function __construct(CoOption $options, $mh) |
|
28
|
30 |
|
{ |
|
29
|
30 |
|
$this->mh = $mh; |
|
30
|
30 |
|
$this->options = $options; |
|
31
|
30 |
|
$this->counter = new ConnectionCounter($options); |
|
32
|
30 |
|
} |
|
33
|
|
|
|
|
34
|
|
|
/** |
|
35
|
|
|
* Call curl_multi_add_handle() or push into queue. |
|
36
|
|
|
* @param resource $ch |
|
37
|
|
|
* @param Deferred $deferred |
|
38
|
|
|
*/ |
|
39
|
10 |
|
public function add($ch, Deferred $deferred = null) |
|
40
|
10 |
|
{ |
|
41
|
10 |
|
$this->counter->isPoolFilled($ch) |
|
42
|
2 |
|
? $this->addReserved($ch, $deferred) |
|
43
|
10 |
|
: $this->addImmediate($ch, $deferred); |
|
44
|
10 |
|
} |
|
45
|
|
|
|
|
46
|
|
|
/** |
|
47
|
|
|
* Are there no cURL handles? |
|
48
|
|
|
* @return bool |
|
49
|
|
|
*/ |
|
50
|
21 |
|
public function isEmpty() |
|
51
|
21 |
|
{ |
|
52
|
21 |
|
return !$this->added && !$this->queue; |
|
53
|
|
|
} |
|
54
|
|
|
|
|
55
|
|
|
/** |
|
56
|
|
|
* Call curl_multi_add_handle(). |
|
57
|
|
|
* @param resource $ch |
|
58
|
|
|
* @param Deferred $deferred |
|
59
|
|
|
*/ |
|
60
|
10 |
View Code Duplication |
private function addImmediate($ch, Deferred $deferred = null) |
|
|
|
|
|
|
61
|
10 |
|
{ |
|
62
|
10 |
|
$errno = curl_multi_add_handle($this->mh, $ch); |
|
63
|
10 |
|
if ($errno !== CURLM_OK) { |
|
64
|
|
|
// @codeCoverageIgnoreStart |
|
65
|
|
|
$msg = curl_multi_strerror($errno) . ": $ch"; |
|
66
|
|
|
$deferred && $deferred->reject(new \RuntimeException($msg)); |
|
67
|
|
|
return; |
|
68
|
|
|
// @codeCoverageIgnoreEnd |
|
69
|
|
|
} |
|
70
|
10 |
|
$this->added[(string)$ch] = $ch; |
|
71
|
10 |
|
$this->counter->addDestination($ch); |
|
72
|
10 |
|
$deferred && $this->deferreds[(string)$ch] = $deferred; |
|
73
|
10 |
|
} |
|
74
|
|
|
|
|
75
|
|
|
/** |
|
76
|
|
|
* Push into queue. |
|
77
|
|
|
* @param resource $ch |
|
78
|
|
|
* @param Deferred $deferred |
|
79
|
|
|
*/ |
|
80
|
2 |
|
private function addReserved($ch, Deferred $deferred = null) |
|
81
|
2 |
|
{ |
|
82
|
2 |
|
$this->queue[(string)$ch] = $ch; |
|
83
|
2 |
|
$deferred && $this->deferreds[(string)$ch] = $deferred; |
|
84
|
2 |
|
} |
|
85
|
|
|
|
|
86
|
|
|
/** |
|
87
|
|
|
* Add cURL handles from waiting queue. |
|
88
|
|
|
* @param array $entry |
|
89
|
|
|
*/ |
|
90
|
10 |
|
protected function interruptConsume(array $entry) |
|
91
|
10 |
|
{ |
|
92
|
10 |
|
$this->counter->removeDestination($entry['handle']); |
|
93
|
10 |
|
if ($this->queue) { |
|
94
|
2 |
|
$this->add(array_shift($this->queue)); |
|
95
|
|
|
} |
|
96
|
10 |
|
} |
|
97
|
|
|
} |
|
98
|
|
|
|
mpyw /
co
Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation.
You can also find more detailed suggestions in the “Code” section of your repository.