blob: c5b80a7f57f0f253742fa362d31158db78c463cd (
plain)
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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
|
<?php
namespace Guzzle\Http\Curl;
use Guzzle\Common\AbstractHasDispatcher;
use Guzzle\Http\Message\RequestInterface;
/**
* Proxies requests and connections to a pool of internal curl_multi handles. Each recursive call will add requests
* to the next available CurlMulti handle.
*/
class CurlMultiProxy extends AbstractHasDispatcher implements CurlMultiInterface
{
protected $handles = array();
protected $groups = array();
protected $queued = array();
protected $maxHandles;
protected $selectTimeout;
/**
* @param int $maxHandles The maximum number of idle CurlMulti handles to allow to remain open
* @param float $selectTimeout timeout for curl_multi_select
*/
public function __construct($maxHandles = 3, $selectTimeout = 1.0)
{
$this->maxHandles = $maxHandles;
$this->selectTimeout = $selectTimeout;
// You can get some weird "Too many open files" errors when sending a large amount of requests in parallel.
// These two statements autoload classes before a system runs out of file descriptors so that you can get back
// valuable error messages if you run out.
class_exists('Guzzle\Http\Message\Response');
class_exists('Guzzle\Http\Exception\CurlException');
}
public function add(RequestInterface $request)
{
$this->queued[] = $request;
return $this;
}
public function all()
{
$requests = $this->queued;
foreach ($this->handles as $handle) {
$requests = array_merge($requests, $handle->all());
}
return $requests;
}
public function remove(RequestInterface $request)
{
foreach ($this->queued as $i => $r) {
if ($request === $r) {
unset($this->queued[$i]);
return true;
}
}
foreach ($this->handles as $handle) {
if ($handle->remove($request)) {
return true;
}
}
return false;
}
public function reset($hard = false)
{
$this->queued = array();
$this->groups = array();
foreach ($this->handles as $handle) {
$handle->reset();
}
if ($hard) {
$this->handles = array();
}
return $this;
}
public function send()
{
if ($this->queued) {
$group = $this->getAvailableHandle();
// Add this handle to a list of handles than is claimed
$this->groups[] = $group;
while ($request = array_shift($this->queued)) {
$group->add($request);
}
try {
$group->send();
array_pop($this->groups);
$this->cleanupHandles();
} catch (\Exception $e) {
// Remove the group and cleanup if an exception was encountered and no more requests in group
if (!$group->count()) {
array_pop($this->groups);
$this->cleanupHandles();
}
throw $e;
}
}
}
public function count()
{
return count($this->all());
}
/**
* Get an existing available CurlMulti handle or create a new one
*
* @return CurlMulti
*/
protected function getAvailableHandle()
{
// Grab a handle that is not claimed
foreach ($this->handles as $h) {
if (!in_array($h, $this->groups, true)) {
return $h;
}
}
// All are claimed, so create one
$handle = new CurlMulti($this->selectTimeout);
$handle->setEventDispatcher($this->getEventDispatcher());
$this->handles[] = $handle;
return $handle;
}
/**
* Trims down unused CurlMulti handles to limit the number of open connections
*/
protected function cleanupHandles()
{
if ($diff = max(0, count($this->handles) - $this->maxHandles)) {
for ($i = count($this->handles) - 1; $i > 0 && $diff > 0; $i--) {
if (!count($this->handles[$i])) {
unset($this->handles[$i]);
$diff--;
}
}
$this->handles = array_values($this->handles);
}
}
}
|
