This commit is contained in:
2025-10-13 09:04:00 +00:00
parent 7a3b2960f8
commit 4ee3212899
230 changed files with 40880 additions and 245 deletions
+349
View File
@@ -0,0 +1,349 @@
<?php
/**
* This file is part of workerman.
*
* Licensed under The MIT License
* For full copyright and license information, please see the MIT-LICENSE.txt
* Redistributions of files must retain the above copyright notice.
*
* @author walkor<walkor@workerman.net>
* @copyright walkor<walkor@workerman.net>
* @link http://www.workerman.net/
* @license http://www.opensource.org/licenses/mit-license.php MIT License
*/
namespace Workerman\Http;
use Revolt\EventLoop;
use RuntimeException;
use Throwable;
use Workerman\Timer;
/**
* Class Http\Client
* @package Workerman\Http
*/
#[\AllowDynamicProperties]
class Client
{
/**
*
*[
* address=>[
* [
* 'url'=>x,
* 'address'=>x
* 'options'=>['method', 'data'=>x, 'success'=>callback, 'error'=>callback, 'headers'=>[..], 'version'=>1.1]
* ],
* ..
* ],
* ..
* ]
* @var array
*/
protected $_queue = array();
/**
* @var array
*/
protected $_connectionPool = null;
/**
* Client constructor.
* @param array $options
*/
public function __construct($options = [])
{
$this->_connectionPool = new ConnectionPool($options);
$this->_connectionPool->on('idle', array($this, 'process'));
}
/**
* Request.
*
* @param $url string
* @param array $options ['method'=>'get', 'data'=>x, 'success'=>callback, 'error'=>callback, 'headers'=>[..], 'version'=>1.1]
* @return mixed|Response
* @throws Throwable
*/
public function request($url, $options = [])
{
$options['url'] = $url;
$needSuspend = !isset($options['success']) && class_exists(EventLoop::class, false);
try {
$address = $this->parseAddress($url);
$this->queuePush($address, ['url' => $url, 'address' => $address, 'options' => &$options]);
$this->process($address);
} catch (Throwable $exception) {
$this->deferError($options, $exception);
return;
}
if ($needSuspend) {
$suspension = EventLoop::getSuspension();
$options['success'] = function ($response) use ($suspension) {
$suspension->resume($response);
};
$options['error'] = function ($response) use ($suspension) {
$suspension->throw($response);
};
return $suspension->suspend();
}
}
/**
* Get.
*
* @param $url
* @param null $success_callback
* @param null $error_callback
* @return mixed|Response
* @throws Throwable
*/
public function get($url, $success_callback = null, $error_callback = null)
{
$options = [];
if ($success_callback) {
$options['success'] = $success_callback;
}
if ($error_callback) {
$options['error'] = $error_callback;
}
return $this->request($url, $options);
}
/**
* Post.
*
* @param $url
* @param array $data
* @param null $success_callback
* @param null $error_callback
* @return mixed|Response
* @throws Throwable
*/
public function post($url, $data = [], $success_callback = null, $error_callback = null)
{
$options = [];
if ($data) {
$options['data'] = $data;
}
if ($success_callback) {
$options['success'] = $success_callback;
}
if ($error_callback) {
$options['error'] = $error_callback;
}
$options['method'] = 'POST';
return $this->request($url, $options);
}
/**
* Process.
* User should not call this.
*
* @param $address
* @return void
* @throws \Exception
*/
public function process($address)
{
$task = $this->queueCurrent($address);
if (!$task) {
return;
}
$url = $task['url'];
$address = $task['address'];
$connection = $this->_connectionPool->fetch($address, strpos($url, 'https') === 0, $task['options']['proxy'] ?? '');
// No connection is in idle state then wait.
if (!$connection) {
return;
}
$connection->errorHandler = function(Throwable $exception) use ($task) {
$this->deferError($task['options'], $exception);
};
$this->queuePop($address);
$options = $task['options'];
$request = new Request($url);
$data = isset($options['data']) ? $options['data'] : '';
if ($data || $data === '0' || $data === 0) {
$method = isset($options['method']) ? strtoupper($options['method']) : null;
if ($method && in_array($method, ['POST', 'PUT', 'PATCH', 'DELETE'])) {
$request->write($options['data']);
} else {
$options['query'] = $data;
}
}
$request->setOptions($options)->attachConnection($connection);
$client = $this;
$request->once('success', function($response) use ($task, $client, $request) {
$client->recycleConnectionFromRequest($request, $response);
try {
$new_request = Request::redirect($request, $response);
} catch (\Exception $exception) {
$this->deferError($task['options'], $exception);
return;
}
// No redirect.
if (!$new_request) {
if (!empty($task['options']['success'])) {
call_user_func($task['options']['success'], $response);
}
return;
}
// Redirect.
$uri = $new_request->getUri();
$url = (string)$uri;
$options = $new_request->getOptions();
// According to RFC 7231, for HTTP status codes 301, 302, or 303, the client should switch the request
// method to GET and remove any payload data
if (in_array($response->getStatusCode(), [301, 302, 303])) {
$options['method'] = 'GET';
$options['data'] = NULL;
}
$address = $this->parseAddress($url);
$task = [
'url' => $url,
'options' => $options,
'address' => $address
];
$this->queueUnshift($address, $task);
$this->process($address);
})->once('error', function($exception) use ($task, $client, $request) {
$client->recycleConnectionFromRequest($request);
$this->deferError($task['options'], $exception);
});
if (isset($options['progress'])) {
$request->on('progress', $options['progress']);
}
$state = $connection->getStatus(false);
if ($state === 'CLOSING' || $state === 'CLOSED') {
$connection->reconnect();
}
$state = $connection->getStatus(false);
if ($state === 'CLOSED' || $state === 'CLOSING') {
return;
}
$request->end('');
}
/**
* Recycle connection from request.
*
* @param $request Request
* @param $response Response
*/
public function recycleConnectionFromRequest($request, $response = null)
{
$connection = $request->getConnection();
if (!$connection) {
return;
}
$connection->onConnect = $connection->onClose = $connection->onMessage = $connection->onError = null;
$request_header_connection = strtolower($request->getHeaderLine('Connection'));
$response_header_connection = $response ? strtolower($response->getHeaderLine('Connection')) : '';
// Close Connection without header Connection: keep-alive
if ('keep-alive' !== $request_header_connection || 'keep-alive' !== $response_header_connection || $request->getProtocolVersion() !== '1.1') {
$connection->close();
}
$request->detachConnection($connection);
$this->_connectionPool->recycle($connection);
}
/**
* Parse address from url.
*
* @param $url
* @return string
*/
protected function parseAddress($url)
{
$info = parse_url($url);
if (empty($info) || !isset($info['host'])) {
throw new RuntimeException("invalid url: $url");
}
$port = isset($info['port']) ? $info['port'] : (strpos($url, 'https') === 0 ? 443 : 80);
return "tcp://{$info['host']}:{$port}";
}
/**
* Queue push.
*
* @param $address
* @param $task
*/
protected function queuePush($address, $task)
{
if (!isset($this->_queue[$address])) {
$this->_queue[$address] = [];
}
$this->_queue[$address][] = $task;
}
/**
* Queue unshift.
*
* @param $address
* @param $task
*/
protected function queueUnshift($address, $task)
{
if (!isset($this->_queue[$address])) {
$this->_queue[$address] = [];
}
$this->_queue[$address] += [$task];
}
/**
* Queue current item.
*
* @param $address
* @return mixed|null
*/
protected function queueCurrent($address)
{
if (empty($this->_queue[$address])) {
return null;
}
reset($this->_queue[$address]);
return current($this->_queue[$address]);
}
/**
* Queue pop.
*
* @param $address
*/
protected function queuePop($address)
{
unset($this->_queue[$address][key($this->_queue[$address])]);
if (empty($this->_queue[$address])) {
unset($this->_queue[$address]);
}
}
/**
* @param $callback
* @param $exception
* @return void
*/
protected function deferError($options, $exception)
{
if (isset($options['error'])) {
Timer::add(0.000001, $options['error'], [$exception], false);
return;
}
$needSuspend = !isset($options['success']) && class_exists(EventLoop::class, false);
if ($needSuspend) {
throw $exception;
}
}
}
+258
View File
@@ -0,0 +1,258 @@
<?php
/**
* This file is part of workerman.
*
* Licensed under The MIT License
* For full copyright and license information, please see the MIT-LICENSE.txt
* Redistributions of files must retain the above copyright notice.
*
* @author walkor<walkor@workerman.net>
* @copyright walkor<walkor@workerman.net>
* @link http://www.workerman.net/
* @license http://www.opensource.org/licenses/mit-license.php MIT License
*/
namespace Workerman\Http;
use \Workerman\Connection\AsyncTcpConnection;
use \Workerman\Timer;
use \Workerman\Worker;
/**
* Class ConnectionPool
* @package Workerman\Http
*/
class ConnectionPool extends Emitter
{
/**
* @var array
*/
protected $_idle = [];
/**
* @var array
*/
protected $_using = [];
/**
* @var int
*/
protected $_timer = 0;
/**
* @var array
*/
protected $_options = [
'max_conn_per_addr' => 128,
'keepalive_timeout' => 15,
'connect_timeout' => 30,
'timeout' => 30,
];
/**
* ConnectionPool constructor.
*
* @param array $option
*/
public function __construct($option = [])
{
$this->_options = array_merge($this->_options, $option);
}
/**
* Fetch an idle connection.
*
* @param $address
* @param bool $ssl
* @param string $proxy
* @return mixed
*/
public function fetch($address, $ssl = false, $proxy = '')
{
$max_con = $this->_options['max_conn_per_addr'];
$targetAddress = $address;
$address = ProxyHelper::addressKey($address, $proxy);
if (!empty($this->_using[$address])) {
if (count($this->_using[$address]) >= $max_con) {
return;
}
}
if (empty($this->_idle[$address])) {
$connection = $this->create($targetAddress, $ssl, $proxy);
$this->_idle[$address][$connection->id] = $connection;
}
$connection = array_pop($this->_idle[$address]);
if (!isset($this->_using[$address])) {
$this->_using[$address] = [];
}
$this->_using[$address][$connection->id] = $connection;
$connection->pool['request_time'] = time();
$this->tryToCreateConnectionCheckTimer();
return $connection;
}
/**
* Recycle an connection.
*
* @param $connection AsyncTcpConnection
*/
public function recycle($connection)
{
$connection_id = $connection->id;
$address = $connection->address;
unset($this->_using[$address][$connection_id]);
if (empty($this->_using[$address])) {
unset($this->_using[$address]);
}
if ($connection->getStatus(false) === 'ESTABLISHED') {
$this->_idle[$address][$connection_id] = $connection;
$connection->pool['idle_time'] = time();
$connection->onConnect = $connection->onMessage = $connection->onError =
$connection->onClose = $connection->onBufferFull = $connection->onBufferDrain = null;
}
$this->tryToCreateConnectionCheckTimer();
$this->emit('idle', $address);
}
/**
* Delete a connection.
*
* @param $connection
*/
public function delete($connection)
{
$connection_id = $connection->id;
$address = $connection->address;
unset($this->_idle[$address][$connection_id]);
if (empty($this->_idle[$address])) {
unset($this->_idle[$address]);
}
unset($this->_using[$address][$connection_id]);
if (empty($this->_using[$address])) {
unset($this->_using[$address]);
}
}
/**
* Close timeout connection.
*/
public function closeTimeoutConnection()
{
if (empty($this->_idle) && empty($this->_using)) {
Timer::del($this->_timer);
$this->_timer = 0;
return;
}
$time = time();
$keepalive_timeout = $this->_options['keepalive_timeout'];
foreach ($this->_idle as $address => $connections) {
if (empty($connections)) {
unset($this->_idle[$address]);
continue;
}
foreach ($connections as $connection) {
if ($time - $connection->pool['idle_time'] >= $keepalive_timeout) {
$this->delete($connection);
$connection->close();
}
}
}
$connect_timeout = $this->_options['connect_timeout'];
$timeout = $this->_options['timeout'];
foreach ($this->_using as $address => $connections) {
if (empty($connections)) {
unset($this->_using[$address]);
continue;
}
foreach ($connections as $connection) {
$state = $connection->getStatus(false);
if ($state === 'CONNECTING') {
$diff = $time - $connection->pool['connect_time'];
if ($diff >= $connect_timeout) {
$connection->onClose = null;
if ($connection->onError) {
try {
call_user_func($connection->onError, $connection, 1, 'connect ' . $connection->getRemoteAddress() . ' timeout after ' . $diff . ' seconds');
} catch (\Throwable $exception) {
$this->delete($connection);
$connection->close();
throw $exception;
}
}
$this->delete($connection);
$connection->close();
}
} elseif ($state === 'ESTABLISHED') {
$diff = $time - $connection->pool['request_time'];
if ($diff >= $timeout) {
if ($connection->onError) {
try {
call_user_func($connection->onError, $connection, 128, 'read ' . $connection->getRemoteAddress() . ' timeout after ' . $diff . ' seconds');
} catch (\Throwable $exception) {
$this->delete($connection);
$connection->close();
throw $exception;
}
}
$this->delete($connection);
$connection->close();
}
}
}
}
gc_collect_cycles();
}
/**
* Create a connection.
*
* @param $address
* @param bool $ssl
* @param string $proxy
* @return AsyncTcpConnection
*/
protected function create($address, $ssl = false, $proxy = '')
{
$context = array(
'ssl' => array(
'verify_peer' => false,
'verify_peer_name' => false,
'allow_self_signed' => true
),
'http' => array(
'proxy' => $proxy
),
);
if (!empty( $this->_options['context'])) {
$context = $this->_options['context'];
}
if (!$ssl) {
unset($context['ssl']);
}
if (empty($proxy)) {
unset($context['http']['proxy']);
}
if (!class_exists(Worker::class) || is_null(Worker::$globalEvent)) {
throw new \Exception('Only the workerman environment is supported.');
}
$connection = new AsyncTcpConnection($address, $context);
if ($ssl) {
$connection->transport = 'ssl';
}
ProxyHelper::setConnectionProxy($connection, $context);
$connection->address = ProxyHelper::addressKey($address, $proxy);
$connection->connect();
$connection->pool = ['connect_time' => time()];
return $connection;
}
/**
* Create Timer.
*/
protected function tryToCreateConnectionCheckTimer()
{
if (!$this->_timer) {
$this->_timer = Timer::add(1, [$this, 'closeTimeoutConnection']);
}
}
}
+134
View File
@@ -0,0 +1,134 @@
<?php
namespace Workerman\Http;
class Emitter
{
/**
* [event=>[[listener1, once?], [listener2,once?], ..], ..]
*/
protected $_eventListenerMap = array();
/**
* On.
*
* @param $event_name
* @param $listener
* @return $this
*/
public function on($event_name, $listener)
{
$this->emit('newListener', $event_name, $listener);
$this->_eventListenerMap[$event_name][] = array($listener, 0);
return $this;
}
/**
* Once.
*
* @param $event_name
* @param $listener
* @return $this
*/
public function once($event_name, $listener)
{
$this->_eventListenerMap[$event_name][] = array($listener, 1);
return $this;
}
/**
* RemoveListener.
*
* @param $event_name
* @param $listener
* @return $this
*/
public function removeListener($event_name, $listener)
{
if(!isset($this->_eventListenerMap[$event_name]))
{
return $this;
}
foreach($this->_eventListenerMap[$event_name] as $key=>$item)
{
if($item[0] === $listener)
{
$this->emit('removeListener', $event_name, $listener);
unset($this->_eventListenerMap[$event_name][$key]);
}
}
if(empty($this->_eventListenerMap[$event_name]))
{
unset($this->_eventListenerMap[$event_name]);
}
return $this;
}
/**
* RemoveAllListeners.
*
* @param null $event_name
* @return $this
*/
public function removeAllListeners($event_name = null)
{
$this->emit('removeListener', $event_name);
if(null === $event_name)
{
$this->_eventListenerMap = array();
return $this;
}
unset($this->_eventListenerMap[$event_name]);
return $this;
}
/**
*
* Listeners.
*
* @param $event_name
* @return array
*/
public function listeners($event_name)
{
if(empty($this->_eventListenerMap[$event_name]))
{
return array();
}
$listeners = array();
foreach($this->_eventListenerMap[$event_name] as $item)
{
$listeners[] = $item[0];
}
return $listeners;
}
/**
* Emit.
*
* @param null $event_name
* @return bool
*/
public function emit($event_name = null)
{
if(empty($event_name) || empty($this->_eventListenerMap[$event_name]))
{
return false;
}
foreach($this->_eventListenerMap[$event_name] as $key=>$item)
{
$args = func_get_args();
unset($args[0]);
call_user_func_array($item[0], $args);
// once ?
if($item[1])
{
unset($this->_eventListenerMap[$event_name][$key]);
if(empty($this->_eventListenerMap[$event_name]))
{
unset($this->_eventListenerMap[$event_name]);
}
}
}
return true;
}
}
+89
View File
@@ -0,0 +1,89 @@
<?php
namespace Workerman\Http;
use Revolt\EventLoop;
use Throwable;
use Workerman\Http\Response;
use Workerman\Http\Client;
/**
* parallel client request
*/
#[\AllowDynamicProperties]
class ParallelClient extends Client
{
protected $_buffer_queues = [];
public function push(string $url, array $options = [])
{
$this->_buffer_queues[] = [$url, $options];
}
public function batch(array $set)
{
$this->_buffer_queues = array_merge($this->_buffer_queues, $set);
}
public function await(bool $errorThrow = false): array
{
if(!class_exists(EventLoop::class, false)) {
throw new \RuntimeException('Please install revolt/event-loop to use parallel client.');
}
$queues = $this->_buffer_queues;
$result = [];
$suspensionArr = array_fill(0, count($queues), EventLoop::getSuspension());
foreach ($queues as $index => $each) {
$suspension = $suspensionArr[$index];
$options = $each[1];
$options['success'] = function ($response) use (&$result, &$suspension, $options, $index) {
$result[$index] = [true, $response];
$suspension->resume();
// custom callback
if (!empty($options['success'])) {
call_user_func($options['success'], $response);
}
};
$options['error'] = function ($exception) use (&$result, &$suspension, $errorThrow, $options, $index) {
$result[$index] = [false, $exception];
try {
if ($errorThrow) {
$suspension->throw($exception);
} else {
$suspension->resume();
}
} catch (Throwable $e) {
unset($suspension);
}
// custom callback
if (!empty($options['error'])) {
call_user_func($options['error'], $exception);
}
};
$this->request($each[0], $options);
}
foreach ($suspensionArr as $index => $suspension) {
$suspension->suspend();
}
ksort($result);
return $result;
}
#[\Override]
protected function deferError($options, $exception)
{
if (!empty($options['error'])) {
call_user_func($options['error'], $exception);
}
}
}
+32
View File
@@ -0,0 +1,32 @@
<?php
namespace Workerman\Http;
use Workerman\Connection\AsyncTcpConnection;
class ProxyHelper
{
public static function setConnectionProxy(AsyncTcpConnection &$connection, array $context): void
{
$httpProxy = $context['http']['proxy'] ?? '';
if (!empty($httpProxy)) {
$proxyScheme = parse_url($httpProxy, PHP_URL_SCHEME);
$proxyString = explode('//', $httpProxy)[1] ?? '';
if ($proxyScheme === 'socks5') {
$connection->proxySocks5 = $proxyString;
} else if ($proxyScheme === 'http') {
$connection->proxyHttp = $proxyString;
}
}
}
public static function addressKey(string $address, string $proxyString): string
{
if (strpos($proxyString, '://') === false) {
return $address;
} else {
$proxyString = explode('//', $proxyString)[1] ?? '';
return $proxyString;
}
}
}
+591
View File
@@ -0,0 +1,591 @@
<?php
/**
* This file is part of workerman.
*
* Licensed under The MIT License
* For full copyright and license information, please see the MIT-LICENSE.txt
* Redistributions of files must retain the above copyright notice.
*
* @author walkor<walkor@workerman.net>
* @copyright walkor<walkor@workerman.net>
* @link http://www.workerman.net/
* @license http://www.opensource.org/licenses/mit-license.php MIT License
*/
namespace Workerman\Http;
use \Workerman\Connection\AsyncTcpConnection;
use \Workerman\Timer;
use Workerman\Psr7\Uri;
/**
* Class Request
* @package Workerman\Http
*/
#[\AllowDynamicProperties]
class Request extends \Workerman\Psr7\Request
{
/**
* @var AsyncTcpConnection
*/
protected $_connection = null;
/**
* @var Emitter
*/
protected $_emitter = null;
/**
* @var Response
*/
protected $_response = null;
/**
* @var string
*/
protected $_recvBuffer = '';
/**
* @var int
*/
protected $_expectedLength = 0;
/**
* @var int
*/
protected $_chunkedLength = 0;
/**
* @var string
*/
protected $_chunkedData = '';
/**
* @var bool
*/
protected $_writeable = true;
/**
* @var bool
*/
protected $_selfConnection = false;
/**
* @var array
*/
protected $_options = [
'allow_redirects' => [
'max' => 5
]
];
/**
* Request constructor.
* @param string $url
*/
public function __construct($url)
{
$this->_emitter = new Emitter();
$headers = [
'User-Agent' => 'workerman/http-client',
'Connection' => 'keep-alive'
];
parent::__construct('GET', $url, $headers, '', '1.1');
}
/**
* @param $options
* @return $this
*/
public function setOptions($options)
{
$this->_options = array_merge($this->_options, $options);
return $this;
}
/**
* @return array
*/
public function getOptions()
{
return $this->_options;
}
/**
* @param $event
* @param $callback
* @return $this
*/
public function on($event, $callback)
{
$this->_emitter->on($event, $callback);
return $this;
}
/**
* @param $event
* @param $callback
* @return $this
*/
public function once($event, $callback)
{
$this->_emitter->once($event, $callback);
return $this;
}
/**
* @param $event
*/
public function emit($event)
{
$args = func_get_args();
call_user_func_array(array($this->_emitter, 'emit'), $args);
}
/**
* @param $event
* @param $listener
* @return $this
*/
public function removeListener($event, $listener)
{
$this->_emitter->removeListener($event, $listener);
return $this;
}
/**
* @param null $event
* @return $this
*/
public function removeAllListeners($event = null)
{
$this->_emitter->removeAllListeners($event);
return $this;
}
/**
* @param $event
* @return $this
*/
public function listeners($event)
{
$this->_emitter->listeners($event);
return $this;
}
/**
* Connect.
*/
protected function connect()
{
$host = $this->getUri()->getHost();
$port = $this->getUri()->getPort();
if (!$port) {
$port = $this->getDefaultPort();
}
$context = array();
if (!empty( $this->_options['context'])) {
$context = $this->_options['context'];
}
$ssl = $this->getUri()->getScheme() === 'https';
if (!$ssl) {
unset($context['ssl']);
}
$connection = new AsyncTcpConnection("tcp://$host:$port", $context);
if ($ssl) {
$connection->transport = 'ssl';
}
ProxyHelper::setConnectionProxy($connection, $context);
$this->attachConnection($connection);
$this->_selfConnection = true;
$connection->connect();
}
/**
* @param string $data
* @return $this
*/
public function write($data = '')
{
if (!$this->writeable()) {
$this->emitError(new \Exception('Request pending and can not send request again'));
return $this;
}
if (empty($data) && $data !== '0' && $data !== 0) {
return $this;
}
if (is_array($data)) {
if (isset($data['multipart'])) {
$multipart = new \Workerman\Psr7\MultipartStream($data['multipart']);
$this->withHeader('Content-Type', 'multipart/form-data; boundary=' . $multipart->getBoundary());
$data = $multipart;
} else {
$data = http_build_query($data, '', '&');
}
}
$this->getBody()->write($data);
return $this;
}
/**
* @return void
*/
public function writeToResponse($buffer)
{
$this->emit('progress', $buffer);
$this->_response->getBody()->write($buffer);
}
/**
* @param string $data
* @throws \Exception
*/
public function end($data = '')
{
if (isset($this->_options['version'])) {
$this->withProtocolVersion($this->_options['version']);
}
if (isset($this->_options['method'])) {
$this->withMethod($this->_options['method']);
}
if (isset($this->_options['headers'])) {
foreach ($this->_options['headers'] as $key => $value) {
$this->withHeader($key, $value);
}
}
$query = isset($this->_options['query']) ? $this->_options['query'] : '';
if ($query || $query === '0') {
if (is_array($query)) {
$query = http_build_query($query, '', '&', PHP_QUERY_RFC3986);
}
$uri = $this->getUri()->withQuery($query);
$this->withUri($uri);
}
if ($data !== '') {
$this->write($data);
}
if ((($data || $data === '0' || $data === 0) || $this->getBody()->getSize()) && !$this->hasHeader('Content-Type')) {
$this->withHeader('Content-Type', 'application/x-www-form-urlencoded');
}
if (!$this->_connection) {
$this->connect();
} else {
if ($this->_connection->getStatus(false) === 'CONNECTING') {
$this->_connection->onConnect = array($this, 'onConnect');
return;
}
$this->doSend();
}
}
/**
* @return bool
*/
public function writeable()
{
return $this->_writeable;
}
public function doSend()
{
if (!$this->writeable()) {
$this->emitError(new \Exception('Request pending and can not send request again'));
return;
}
$this->_writeable = false;
$body_size = $this->getBody()->getSize();
if ($body_size) {
$this->withHeaders(['Content-Length' => $body_size]);
}
$package = \Workerman\Psr7\str($this);
$this->_connection->send($package);
}
public function onConnect()
{
try {
$this->doSend();
} catch (\Exception $e) {
$this->emitError($e);
}
}
/**
* @param $connection
* @param $recv_buffer
*/
public function onMessage($connection, $recv_buffer)
{
try {
$this->_recvBuffer .= $recv_buffer;
if (!strpos($this->_recvBuffer, "\r\n\r\n")) {
return;
}
$response_data = \Workerman\Psr7\_parse_message($this->_recvBuffer);
if (!preg_match('/^HTTP\/.* [0-9]{3}( .*|$)/', $response_data['start-line'])) {
throw new \InvalidArgumentException('Invalid response string: ' . $response_data['start-line']);
}
$parts = explode(' ', $response_data['start-line'], 3);
$this->_response = new Response(
$parts[1],
$response_data['headers'],
'',
explode('/', $parts[0])[1],
isset($parts[2]) ? $parts[2] : null
);
$this->checkComplete($response_data['body']);
} catch (\Exception $e) {
$this->emitError($e);
}
}
/**
* @param $body
*/
protected function checkComplete($body)
{
$status_code = $this->_response->getStatusCode();
$content_length = $this->_response->getHeaderLine('Content-Length');
if ($content_length === '0' || ($status_code >= 100 && $status_code < 200)
|| $status_code === 204 || $status_code === 304) {
$this->emitSuccess();
return;
}
$transfer_encoding = $this->_response->getHeaderLine('Transfer-Encoding');
// Chunked
if ($transfer_encoding && false === strpos($transfer_encoding, 'identity')) {
$this->_connection->onMessage = array($this, 'handleChunkedData');
$this->handleChunkedData($this->_connection, $body);
} else {
$this->_connection->onMessage = array($this, 'handleData');
$content_length = (int)$this->_response->getHeaderLine('Content-Length');
if (!$content_length) {
// Wait close
$this->_connection->onClose = array($this, 'emitSuccess');
} else {
$this->_expectedLength = $content_length;
}
$this->handleData($this->_connection, $body);
}
}
/**
* @param $connection
* @param $data
*/
public function handleData($connection, $data)
{
try {
$body = $this->_response->getBody();
$this->writeToResponse($data);
if ($this->_expectedLength) {
$recv_length = $body->getSize();
if ($this->_expectedLength <= $recv_length) {
$this->emitSuccess();
}
}
} catch (\Exception $e) {
$this->emitError($e);
}
}
/**
* @param $connection
* @param $buffer
*/
public function handleChunkedData($connection, $buffer)
{
try {
if ($buffer !== '') {
$this->_chunkedData .= $buffer;
}
$recv_len = strlen($this->_chunkedData);
if ($recv_len < 2) {
return;
}
// Get chunked length
if ($this->_chunkedLength === 0) {
$crlf_position = strpos($this->_chunkedData, "\r\n");
if ($crlf_position === false && strlen($this->_chunkedData) > 1024) {
$this->emitError(new \Exception('bad chunked length'));
return;
}
if ($crlf_position === false) {
return;
}
$length_chunk = substr($this->_chunkedData, 0, $crlf_position);
if (strpos($crlf_position, ';') !== false) {
list($length_chunk) = explode(';', $length_chunk, 2);
}
$length = hexdec(ltrim(trim($length_chunk), "0"));
if ($length === 0) {
$this->emitSuccess();
return;
}
$this->_chunkedLength = $length + 2;
$this->_chunkedData = substr($this->_chunkedData, $crlf_position + 2);
$this->handleChunkedData($connection, '');
return;
}
// Get chunked data
if ($recv_len >= $this->_chunkedLength) {
$this->writeToResponse(substr($this->_chunkedData, 0, $this->_chunkedLength - 2));
$this->_chunkedData = substr($this->_chunkedData, $this->_chunkedLength);
$this->_chunkedLength = 0;
$this->handleChunkedData($connection, '');
}
} catch (\Exception $e) {
$this->emitError($e);
}
}
/**
* onError.
*/
public function onError($connection, $code, $msg)
{
$this->emitError(new \Exception($msg, $code));
}
/**
* emitSuccess.
*/
public function emitSuccess()
{
$this->emit('success', $this->_response);
}
public function emitError($e)
{
try {
$this->emit('error', $e);
} finally {
$this->_connection && $this->_connection->destroy();
}
}
/**
* @param $request Request
* @param $response Response
* @return $this|bool
*/
public static function redirect($request, $response)
{
if (substr($response->getStatusCode(), 0, 1) != '3'
|| !$response->hasHeader('Location')
) {
return false;
}
$options = $request->getOptions();
self::guardMax($options);
$location = \Workerman\Psr7\UriResolver::resolve(
$request->getUri(),
new \Workerman\Psr7\Uri($response->getHeaderLine('Location'))
);
\Workerman\Psr7\rewind_body($request);
$new_request = (new Request($location))->setOptions($options)->withBody($request->getBody());
return $new_request;
}
private static function guardMax(array &$options)
{
$current = isset($options['__redirect_count'])
? $options['__redirect_count']
: 0;
$options['__redirect_count'] = $current + 1;
$max = $options['allow_redirects']['max'];
if ($options['__redirect_count'] > $max) {
throw new \Exception("Too many redirects. will not follow more than {$max} redirects");
}
}
/**
* onUnexpectClose.
*/
public function onUnexpectClose()
{
$this->emitError(new \Exception('The connection to ' . $this->_connection->getRemoteIp() . ' has been closed.'));
}
/**
* @return int
*/
protected function getDefaultPort()
{
return ('https' === $this->getUri()->getScheme()) ? 443 : 80;
}
/**
* detachConnection.
*
* @return void
*/
public function detachConnection()
{
$this->cleanConnection();
// 不是连接池的连接则断开
if ($this->_selfConnection) {
$this->_connection->close();
return;
}
$this->_writeable = true;
}
/**
* @return \Workerman\Connection\AsyncTcpConnection
*/
public function getConnection()
{
return $this->_connection;
}
/**
* attachConnection.
*
* @param $connection \Workerman\Connection\AsyncTcpConnection
* @return $this
*/
public function attachConnection($connection)
{
$connection->onConnect = array($this, 'onConnect');
$connection->onMessage = array($this, 'onMessage');
$connection->onError = array($this, 'onError');
$connection->onClose = array($this, 'onUnexpectClose');
$this->_connection = $connection;
return $this;
}
/**
* cleanConnection.
*/
protected function cleanConnection()
{
$connection = $this->_connection;
$connection->onConnect = $connection->onMessage = $connection->onError =
$connection->onClose = $connection->onBufferFull = $connection->onBufferDrain = null;
$this->_connection = null;
$this->_emitter->removeAllListeners();
}
}
+24
View File
@@ -0,0 +1,24 @@
<?php
/**
* This file is part of workerman.
*
* Licensed under The MIT License
* For full copyright and license information, please see the MIT-LICENSE.txt
* Redistributions of files must retain the above copyright notice.
*
* @author walkor<walkor@workerman.net>
* @copyright walkor<walkor@workerman.net>
* @link http://www.workerman.net/
* @license http://www.opensource.org/licenses/mit-license.php MIT License
*/
namespace Workerman\Http;
/**
* Class Response
* @package Workerman\Http
*/
#[\AllowDynamicProperties]
class Response extends \Workerman\Psr7\Response
{
}