AbstractConsumer.php
2 years ago
CurlConsumer.php
2 years ago
FileConsumer.php
2 years ago
SocketConsumer.php
2 years ago
index.php
2 years ago
CurlConsumer.php
143 lines
| 1 | <?php |
| 2 | if (!defined('ABSPATH')) exit; |
| 3 | require_once(dirname(__FILE__) . "/AbstractConsumer.php"); |
| 4 | class ConsumerStrategies_CurlConsumer extends ConsumerStrategies_AbstractConsumer { |
| 5 | protected $_host; |
| 6 | protected $_endpoint; |
| 7 | protected $_connect_timeout; |
| 8 | protected $_timeout; |
| 9 | protected $_protocol; |
| 10 | protected $_fork = null; |
| 11 | protected $_num_threads; |
| 12 | function __construct($options) { |
| 13 | parent::__construct($options); |
| 14 | $this->_host = $options['host']; |
| 15 | $this->_endpoint = $options['endpoint']; |
| 16 | $this->_connect_timeout = isset($options['connect_timeout']) ? $options['connect_timeout'] : 5; |
| 17 | $this->_timeout = isset($options['timeout']) ? $options['timeout'] : 30; |
| 18 | $this->_protocol = isset($options['use_ssl']) && $options['use_ssl'] == true ? "https" : "http"; |
| 19 | $this->_fork = isset($options['fork']) ? ($options['fork'] == true) : false; |
| 20 | $this->_num_threads = isset($options['num_threads']) ? max(1, intval($options['num_threads'])) : 1; |
| 21 | // ensure the environment is workable for the given settings |
| 22 | if ($this->_fork == true) { |
| 23 | $exists = function_exists('exec'); |
| 24 | if (!$exists) { |
| 25 | throw new Exception('The "exec" function must exist to use the cURL consumer in "fork" mode. Try setting fork = false or use another consumer.'); |
| 26 | } |
| 27 | $disabled = explode(', ', ini_get('disable_functions')); |
| 28 | $enabled = !in_array('exec', $disabled); |
| 29 | if (!$enabled) { |
| 30 | throw new Exception('The "exec" function must be enabled to use the cURL consumer in "fork" mode. Try setting fork = false or use another consumer.'); |
| 31 | } |
| 32 | } else { |
| 33 | if (!function_exists('curl_init')) { |
| 34 | throw new Exception('The cURL PHP extension is required to use the cURL consumer with fork = false. Try setting fork = true or use another consumer.'); |
| 35 | } |
| 36 | } |
| 37 | } |
| 38 | public function persist($batch) { |
| 39 | if (count($batch) > 0) { |
| 40 | $url = $this->_protocol . "://" . $this->_host . $this->_endpoint; |
| 41 | if ($this->_fork) { |
| 42 | $data = "data=" . $this->_encode($batch); |
| 43 | return $this->_execute_forked($url, $data); |
| 44 | } else { |
| 45 | return $this->_execute($url, $batch); |
| 46 | } |
| 47 | } else { |
| 48 | return true; |
| 49 | } |
| 50 | } |
| 51 | protected function _execute($url, $batch) { |
| 52 | if ($this->_debug()) { |
| 53 | $this->_log("Making blocking cURL call to $url"); |
| 54 | } |
| 55 | $mh = curl_multi_init(); |
| 56 | $chs = array(); |
| 57 | $batch_size = ceil(count($batch) / $this->_num_threads); |
| 58 | for ($i=0; $i<$this->_num_threads && !empty($batch); $i++) { |
| 59 | $ch = curl_init(); |
| 60 | $chs[] = $ch; |
| 61 | $data = "data=" . $this->_encode(array_splice($batch, 0, $batch_size)); |
| 62 | curl_setopt($ch, CURLOPT_URL, $url); |
| 63 | curl_setopt($ch, CURLOPT_HEADER, 0); |
| 64 | curl_setopt($ch, CURLOPT_CONNECTTIMEOUT, $this->_connect_timeout); |
| 65 | curl_setopt($ch, CURLOPT_TIMEOUT, $this->_timeout); |
| 66 | curl_setopt($ch, CURLOPT_POST, 1); |
| 67 | curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1); |
| 68 | curl_setopt($ch, CURLOPT_POSTFIELDS, $data); |
| 69 | curl_multi_add_handle($mh,$ch); |
| 70 | } |
| 71 | $running = 0; |
| 72 | do { |
| 73 | curl_multi_exec($mh, $running); |
| 74 | curl_multi_select($mh); |
| 75 | } while ($running > 0); |
| 76 | $info = curl_multi_info_read($mh); |
| 77 | $error = false; |
| 78 | foreach ($chs as $ch) { |
| 79 | $response = curl_multi_getcontent($ch); |
| 80 | if (false === $response) { |
| 81 | $this->_handleError(curl_errno($ch), curl_error($ch)); |
| 82 | $error = true; |
| 83 | } |
| 84 | elseif ("1" != trim($response)) { |
| 85 | $this->_handleError(0, $response); |
| 86 | $error = true; |
| 87 | } |
| 88 | curl_multi_remove_handle($mh, $ch); |
| 89 | } |
| 90 | if (CURLE_OK != $info['result']) { |
| 91 | $this->_handleError($info['result'], "cURL error with code=".$info['result']); |
| 92 | $error = true; |
| 93 | } |
| 94 | curl_multi_close($mh); |
| 95 | return !$error; |
| 96 | } |
| 97 | protected function _execute_forked($url, $data) { |
| 98 | if ($this->_debug()) { |
| 99 | $this->_log("Making forked cURL call to $url"); |
| 100 | } |
| 101 | $exec = 'curl -X POST -H "Content-Type: application/x-www-form-urlencoded" -d ' . $data . ' "' . $url . '"'; |
| 102 | if(!$this->_debug()) { |
| 103 | $exec .= " >/dev/null 2>&1 &"; |
| 104 | } |
| 105 | exec($exec, $output, $return_var); |
| 106 | if ($return_var != 0) { |
| 107 | $this->_handleError($return_var, $output); |
| 108 | } |
| 109 | return $return_var == 0; |
| 110 | } |
| 111 | public function getConnectTimeout() |
| 112 | { |
| 113 | return $this->_connect_timeout; |
| 114 | } |
| 115 | public function getEndpoint() |
| 116 | { |
| 117 | return $this->_endpoint; |
| 118 | } |
| 119 | public function getFork() |
| 120 | { |
| 121 | return $this->_fork; |
| 122 | } |
| 123 | public function getHost() |
| 124 | { |
| 125 | return $this->_host; |
| 126 | } |
| 127 | public function getOptions() |
| 128 | { |
| 129 | return $this->_options; |
| 130 | } |
| 131 | public function getProtocol() |
| 132 | { |
| 133 | return $this->_protocol; |
| 134 | } |
| 135 | public function getTimeout() |
| 136 | { |
| 137 | return $this->_timeout; |
| 138 | } |
| 139 | public function getNumThreads() { |
| 140 | return $this->_num_threads; |
| 141 | } |
| 142 | } |
| 143 |