1 | <?php |
||||
2 | |||||
3 | namespace Pheanstalk; |
||||
4 | |||||
5 | use DOMXPath as DOMXPath; |
||||
6 | use Pheanstalk\Socket\NativeSocket; |
||||
7 | |||||
8 | /** |
||||
9 | * A connection to a beanstalkd server. |
||||
10 | * |
||||
11 | * @author Paul Annesley |
||||
12 | * @package Pheanstalk |
||||
13 | * @license http://www.opensource.org/licenses/mit-license.php |
||||
14 | */ |
||||
15 | class Connection |
||||
16 | { |
||||
17 | const CRLF = "\r\n"; |
||||
18 | const CRLF_LENGTH = 2; |
||||
19 | const DEFAULT_CONNECT_TIMEOUT = 2; |
||||
20 | |||||
21 | // responses which are global errors, mapped to their exception short-names |
||||
22 | private static $_errorResponses = [ |
||||
23 | Response::RESPONSE_OUT_OF_MEMORY => 'OutOfMemory', |
||||
24 | Response::RESPONSE_INTERNAL_ERROR => 'InternalError', |
||||
25 | Response::RESPONSE_DRAINING => 'Draining', |
||||
26 | Response::RESPONSE_BAD_FORMAT => 'BadFormat', |
||||
27 | Response::RESPONSE_UNKNOWN_COMMAND => 'UnknownCommand', |
||||
28 | Response::RESPONSE_WORKFLOW_ALREADY_EXISTS => 'DuplicateEntry', |
||||
29 | Response::RESPONSE_SERVER_ERROR => '', |
||||
30 | ]; |
||||
31 | |||||
32 | // responses which are followed by data |
||||
33 | private static $_dataResponses = [ |
||||
0 ignored issues
–
show
introduced
by
![]() |
|||||
34 | Response::RESPONSE_RESERVED, |
||||
35 | Response::RESPONSE_FOUND, |
||||
36 | Response::RESPONSE_OK, |
||||
37 | ]; |
||||
38 | |||||
39 | private $_socket; |
||||
40 | private $_hostname; |
||||
41 | private $_port; |
||||
42 | private $_connectTimeout; |
||||
43 | private $_connectPersistent; |
||||
44 | protected $user; |
||||
45 | protected $password; |
||||
46 | |||||
47 | /** |
||||
48 | * @param string $hostname |
||||
49 | * @param string $user |
||||
50 | * @param string $password |
||||
51 | * @param int $port |
||||
52 | * @param float $connectTimeout |
||||
53 | * @param bool $connectPersistent |
||||
54 | */ |
||||
55 | 27 | public function __construct($hostname, $user = null, $password = null, $port = 5000, $connectTimeout = null, $connectPersistent = false) |
|||
56 | { |
||||
57 | 27 | if (is_null($connectTimeout) || !is_numeric($connectTimeout)) { |
|||
58 | 24 | $connectTimeout = self::DEFAULT_CONNECT_TIMEOUT; |
|||
59 | } |
||||
60 | |||||
61 | 27 | $this->_hostname = $hostname; |
|||
62 | 27 | $this->_port = $port; |
|||
63 | 27 | $this->user = $user; |
|||
64 | 27 | $this->password = $password; |
|||
65 | 27 | $this->_connectTimeout = $connectTimeout; |
|||
66 | 27 | $this->_connectPersistent = $connectPersistent; |
|||
67 | } |
||||
68 | |||||
69 | /** |
||||
70 | * @return bool |
||||
71 | */ |
||||
72 | 1 | public function hasSocket() |
|||
73 | { |
||||
74 | 1 | return isset($this->_socket); |
|||
75 | } |
||||
76 | |||||
77 | /** |
||||
78 | * Disconnect the socket. |
||||
79 | * Subsequent socket operations will create a new connection. |
||||
80 | */ |
||||
81 | 1 | public function disconnect() |
|||
82 | { |
||||
83 | 1 | $this->getSocket()->disconnect(); |
|||
84 | 1 | $this->_socket = null; |
|||
85 | } |
||||
86 | |||||
87 | /** |
||||
88 | * @param Command $command |
||||
89 | * |
||||
90 | * @throws Exception\ClientException |
||||
91 | * |
||||
92 | * @return mixed |
||||
93 | */ |
||||
94 | 24 | public function dispatchCommand(Command $command) |
|||
95 | { |
||||
96 | 24 | $socket = $this->getSocket(); |
|||
97 | |||||
98 | 23 | $dom = $this->build_query($command->getGroup(), $command->getAction(), $command->getFilters(), $command->getParameters()); |
|||
0 ignored issues
–
show
$command->getAction() of type string is incompatible with the type boolean expected by parameter $action of Pheanstalk\Connection::build_query() .
(
Ignorable by Annotation
)
If this is a false-positive, you can also ignore this issue in your code via the
![]() |
|||||
99 | 23 | $xml = $dom->saveXML(); |
|||
100 | 23 | $socket->write($xml); |
|||
101 | |||||
102 | 23 | $responseLine = $socket->getLine(); |
|||
103 | 23 | $xml = new \SimpleXMLElement($responseLine); |
|||
104 | 23 | $json = json_encode($xml); |
|||
105 | 23 | $responseLine = json_decode($json, true); |
|||
106 | 23 | $responseName = preg_replace('#^(\S+).*$#s', '$1', $responseLine["@attributes"]['status']); |
|||
107 | 23 | if ($responseName === "KO") { |
|||
108 | 5 | $exceptionType = $responseLine['@attributes']['error-code'] ?? Response::RESPONSE_SERVER_ERROR; |
|||
109 | 5 | $exception = sprintf( |
|||
110 | 5 | '\Pheanstalk\Exception\Server%sException', |
|||
111 | 5 | self::$_errorResponses[$exceptionType] ?? '' |
|||
112 | ); |
||||
113 | 5 | throw new $exception(sprintf( |
|||
114 | 5 | "%s while executing %s:%s", |
|||
115 | 5 | $responseLine['@attributes']['error'], |
|||
116 | 5 | $command->getGroup(), |
|||
117 | 5 | $command->getAction() |
|||
118 | )); |
||||
119 | } |
||||
120 | |||||
121 | |||||
122 | 21 | $data = $responseLine; |
|||
123 | |||||
124 | |||||
125 | return $command |
||||
126 | 21 | ->getResponseParser() |
|||
127 | 21 | ->parseResponse($responseLine["@attributes"]['status'], $data); |
|||
128 | } |
||||
129 | |||||
130 | /** |
||||
131 | * Returns the connect timeout for this connection. |
||||
132 | * |
||||
133 | * @return float |
||||
134 | */ |
||||
135 | 1 | public function getConnectTimeout() |
|||
136 | { |
||||
137 | 1 | return $this->_connectTimeout; |
|||
138 | } |
||||
139 | |||||
140 | /** |
||||
141 | * Returns the host for this connection. |
||||
142 | * |
||||
143 | * @return string |
||||
144 | */ |
||||
145 | 1 | public function getHost() |
|||
146 | { |
||||
147 | 1 | return $this->_hostname; |
|||
148 | } |
||||
149 | |||||
150 | /** |
||||
151 | * Returns the port for this connection. |
||||
152 | * |
||||
153 | * @return int |
||||
154 | */ |
||||
155 | 1 | public function getPort() |
|||
156 | { |
||||
157 | 1 | return $this->_port; |
|||
158 | } |
||||
159 | |||||
160 | // ---------------------------------------- |
||||
161 | |||||
162 | /** |
||||
163 | * Socket handle for the connection to beanstalkd. |
||||
164 | * |
||||
165 | * @throws Exception\ConnectionException |
||||
166 | * |
||||
167 | * @return Socket |
||||
168 | */ |
||||
169 | 26 | public function getSocket() |
|||
170 | { |
||||
171 | 26 | if (!isset($this->_socket)) { |
|||
172 | 25 | $this->_socket = new NativeSocket( |
|||
173 | 25 | $this->_hostname, |
|||
174 | 25 | $this->_port, |
|||
175 | 25 | $this->_connectTimeout, |
|||
0 ignored issues
–
show
It seems like
$this->_connectTimeout can also be of type double ; however, parameter $connectTimeout of Pheanstalk\Socket\NativeSocket::__construct() does only seem to accept integer , maybe add an additional type check?
(
Ignorable by Annotation
)
If this is a false-positive, you can also ignore this issue in your code via the
![]() |
|||||
176 | 25 | $this->_connectPersistent |
|||
177 | ); |
||||
178 | 23 | $response = $this->_socket->getLine(); |
|||
179 | 23 | $xml = new \SimpleXMLElement($response); |
|||
180 | 23 | $challenge = (string) $xml['challenge']; |
|||
181 | 23 | if (!empty($challenge)) { |
|||
182 | 23 | $hmac = hash_hmac("sha1", hex2bin($challenge), sha1($this->password, true)); |
|||
183 | 23 | $dom = $this->build_query('auth', false, ["response" => $hmac, "user" => $this->user]); |
|||
184 | 23 | $this->_socket->write($dom->saveXML()); |
|||
185 | 23 | $recv = $this->_socket->getLine(); |
|||
0 ignored issues
–
show
|
|||||
186 | } |
||||
187 | } |
||||
188 | |||||
189 | 24 | return $this->_socket; |
|||
190 | } |
||||
191 | |||||
192 | /** |
||||
193 | * @param NativeSocket $socket |
||||
194 | */ |
||||
195 | 2 | public function setSocket(NativeSocket $socket) |
|||
196 | { |
||||
197 | 2 | $this->_socket = $socket; |
|||
198 | } |
||||
199 | |||||
200 | /** |
||||
201 | * @param $name |
||||
202 | * @param bool $action |
||||
203 | * @param array $attributes |
||||
204 | * @param array $parameters |
||||
205 | * |
||||
206 | * @return \DOMDocument |
||||
207 | */ |
||||
208 | 24 | protected function build_query($name, $action = false, $attributes = [], $parameters = []) |
|||
0 ignored issues
–
show
|
|||||
209 | { |
||||
210 | 24 | $dom = new \DOMDocument("1.0", "utf-8"); |
|||
211 | 24 | $root = $dom->createElement($name); |
|||
212 | 24 | if ($action) { |
|||
213 | 23 | $root->setAttribute('action', $action); |
|||
0 ignored issues
–
show
$action of type true is incompatible with the type string expected by parameter $value of DOMElement::setAttribute() .
(
Ignorable by Annotation
)
If this is a false-positive, you can also ignore this issue in your code via the
![]() |
|||||
214 | } |
||||
215 | 24 | foreach ($attributes as $key => $value) { |
|||
216 | 24 | $root->setAttribute($key, $value); |
|||
217 | } |
||||
218 | 24 | foreach ($parameters as $parameter => $value) { |
|||
219 | 1 | $param = $dom->createElement('parameter'); |
|||
220 | 1 | $param->setAttribute('name', $parameter); |
|||
221 | 1 | $param->setAttribute('value', $value); |
|||
222 | 1 | $root->appendChild($param); |
|||
223 | } |
||||
224 | 24 | $dom->appendChild($root); |
|||
225 | 24 | return $dom; |
|||
226 | } |
||||
227 | |||||
228 | /** |
||||
229 | * Checks connection to the beanstalkd socket. |
||||
230 | * |
||||
231 | * @return true|false |
||||
232 | */ |
||||
233 | 2 | public function isServiceListening() |
|||
234 | { |
||||
235 | try { |
||||
236 | 2 | $this->getSocket(); |
|||
237 | |||||
238 | 1 | return true; |
|||
239 | 1 | } catch (Exception\ConnectionException $e) { |
|||
240 | 1 | return false; |
|||
241 | } |
||||
242 | } |
||||
243 | } |
||||
244 |