PyRowMan /
pheanstalk
| 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 = [ |
||||
| 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
Bug
introduced
by
Loading history...
|
|||||
| 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
Loading history...
|
|||||
| 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 = []) |
|||
| 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
Loading history...
|
|||||
| 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 |