updated requestsqueue
This commit is contained in:
kremsy
parent
e706ca8b29
commit
ec604f7a64
@ -1,235 +1,220 @@
|
|||||||
<?php
|
<?php
|
||||||
|
|
||||||
namespace cURL;
|
namespace cURL;
|
||||||
|
|
||||||
use Symfony\Component\EventDispatcher\EventDispatcher;
|
use Symfony\Component\EventDispatcher\EventDispatcher;
|
||||||
|
|
||||||
class RequestsQueue extends EventDispatcher implements RequestsQueueInterface, \Countable
|
class RequestsQueue extends EventDispatcher implements RequestsQueueInterface, \Countable {
|
||||||
{
|
/**
|
||||||
/**
|
* @var Options Default options for new Requests attached to RequestsQueue
|
||||||
* @var Options Default options for new Requests attached to RequestsQueue
|
*/
|
||||||
*/
|
protected $defaultOptions = null;
|
||||||
protected $defaultOptions = null;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* @var resource cURL multi handler
|
|
||||||
*/
|
|
||||||
protected $mh;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* @var int Amount of requests running
|
|
||||||
*/
|
|
||||||
protected $runningCount = 0;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* @var Request[] Array of requests attached
|
|
||||||
*/
|
|
||||||
protected $queue = array();
|
|
||||||
|
|
||||||
/**
|
|
||||||
* @var array Array of requests running
|
|
||||||
*/
|
|
||||||
protected $running = array();
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Initializes curl_multi handler
|
|
||||||
*/
|
|
||||||
public function __construct()
|
|
||||||
{
|
|
||||||
$this->mh = curl_multi_init();
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Destructor, closes curl_multi handler
|
|
||||||
*
|
|
||||||
* @return void
|
|
||||||
*/
|
|
||||||
public function __destruct()
|
|
||||||
{
|
|
||||||
if (isset($this->mh)) {
|
|
||||||
curl_multi_close($this->mh);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Returns cURL\Options object with default request's options
|
|
||||||
*
|
|
||||||
* @return Options
|
|
||||||
*/
|
|
||||||
public function getDefaultOptions()
|
|
||||||
{
|
|
||||||
if (!isset($this->defaultOptions)) {
|
|
||||||
$this->defaultOptions = new Options();
|
|
||||||
}
|
|
||||||
return $this->defaultOptions;
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Overrides default options with given Options object
|
|
||||||
*
|
|
||||||
* @param Options $defaultOptions New options
|
|
||||||
* @return void
|
|
||||||
*/
|
|
||||||
public function setDefaultOptions(Options $defaultOptions)
|
|
||||||
{
|
|
||||||
$this->defaultOptions = $defaultOptions;
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Get cURL multi handle
|
|
||||||
*
|
|
||||||
* @return resource
|
|
||||||
*/
|
|
||||||
public function getHandle()
|
|
||||||
{
|
|
||||||
return $this->mh;
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Attach request to queue.
|
|
||||||
*
|
|
||||||
* @param Request $request Request to add
|
|
||||||
* @return self
|
|
||||||
*/
|
|
||||||
public function attach(Request $request)
|
|
||||||
{
|
|
||||||
$this->queue[$request->getUID()] = $request;
|
|
||||||
return $this;
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Detach request from queue.
|
|
||||||
*
|
|
||||||
* @param Request $request Request to remove
|
|
||||||
* @return self
|
|
||||||
*/
|
|
||||||
public function detach(Request $request)
|
|
||||||
{
|
|
||||||
unset($this->queue[$request->getUID()]);
|
|
||||||
return $this;
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Processes handles which are ready and removes them from pool.
|
|
||||||
*
|
|
||||||
* @return int Amount of requests completed
|
|
||||||
*/
|
|
||||||
protected function read()
|
|
||||||
{
|
|
||||||
$n = 0;
|
|
||||||
while ($info = curl_multi_info_read($this->mh)) {
|
|
||||||
$n++;
|
|
||||||
$request = $this->queue[(int)$info['handle']];
|
|
||||||
$result = $info['result'];
|
|
||||||
|
|
||||||
curl_multi_remove_handle($this->mh, $request->getHandle());
|
|
||||||
unset($this->running[$request->getUID()]);
|
|
||||||
$this->detach($request);
|
|
||||||
|
|
||||||
$event = new Event();
|
|
||||||
$event->request = $request;
|
|
||||||
$event->response = new Response($request, curl_multi_getcontent($request->getHandle()));
|
|
||||||
if ($result !== CURLE_OK) {
|
|
||||||
$event->response->setError(new Error(curl_error($request->getHandle()), $result));
|
|
||||||
}
|
|
||||||
$event->queue = $this;
|
|
||||||
$this->dispatch('complete', $event);
|
|
||||||
$request->dispatch('complete', $event);
|
|
||||||
}
|
|
||||||
|
|
||||||
return $n;
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Returns count of handles in queue
|
|
||||||
*
|
|
||||||
* @return int Handles count
|
|
||||||
*/
|
|
||||||
public function count()
|
|
||||||
{
|
|
||||||
return count($this->queue);
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Executes requests in parallel
|
|
||||||
*
|
|
||||||
* @return void
|
|
||||||
*/
|
|
||||||
public function send()
|
|
||||||
{
|
|
||||||
while ($this->socketPerform()) {
|
|
||||||
$this->socketSelect();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Returns requests present in $queue but not in $running
|
|
||||||
*
|
|
||||||
* @return Request[] Array of requests
|
|
||||||
*/
|
|
||||||
protected function getRequestsNotRunning()
|
|
||||||
{
|
|
||||||
$map = $this->queue;
|
|
||||||
foreach($this->running as $k => $v) unset($map[$k]);
|
|
||||||
return $map;
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Download available data on socket.
|
* @var resource cURL multi handler
|
||||||
*
|
*/
|
||||||
* @throws Exception
|
protected $mh;
|
||||||
* @return bool TRUE when there are any requests on queue, FALSE when finished
|
|
||||||
*/
|
|
||||||
public function socketPerform()
|
|
||||||
{
|
|
||||||
if ($this->count() == 0) {
|
|
||||||
throw new Exception('Cannot perform if there are no requests in queue.');
|
|
||||||
}
|
|
||||||
|
|
||||||
$notRunning = $this->getRequestsNotRunning();
|
/**
|
||||||
do {
|
* @var Request[] Array of requests attached
|
||||||
/**
|
*/
|
||||||
* Apply cURL options to new requests
|
protected $queue = array();
|
||||||
*/
|
|
||||||
foreach ($notRunning as $request) {
|
|
||||||
$this->getDefaultOptions()->applyTo($request);
|
|
||||||
$request->getOptions()->applyTo($request);
|
|
||||||
curl_multi_add_handle($this->mh, $request->getHandle());
|
|
||||||
$this->running[$request->getUID()] = $request;
|
|
||||||
}
|
|
||||||
|
|
||||||
$runningBefore = $this->runningCount;
|
|
||||||
do {
|
|
||||||
$mrc = curl_multi_exec($this->mh, $this->runningCount);
|
|
||||||
} while ($mrc === CURLM_CALL_MULTI_PERFORM);
|
|
||||||
$runningAfter = $this->runningCount;
|
|
||||||
|
|
||||||
if ($runningAfter < $runningBefore) {
|
/**
|
||||||
$this->read();
|
* @var array Array of requests added to curl multi handle
|
||||||
}
|
*/
|
||||||
|
protected $running = array();
|
||||||
$notRunning = $this->getRequestsNotRunning();
|
|
||||||
} while (count($notRunning) > 0);
|
/**
|
||||||
// Why the loop? New requests might be added at runtime on 'complete' event.
|
* Initializes curl_multi handler
|
||||||
// So we need to attach them to curl_multi handle immediately.
|
*/
|
||||||
|
public function __construct() {
|
||||||
return $this->count() > 0;
|
$this->mh = curl_multi_init();
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Waits until activity on socket
|
* Destructor, closes curl_multi handler
|
||||||
* On success, returns TRUE. On failure, this function will
|
*
|
||||||
* return FALSE on a select failure or timeout (from the underlying
|
* @return void
|
||||||
* select system call)
|
*/
|
||||||
*
|
public function __destruct() {
|
||||||
* @param float|int $timeout Maximum time to wait
|
if (isset($this->mh)) {
|
||||||
* @throws Exception
|
curl_multi_close($this->mh);
|
||||||
* @return bool
|
}
|
||||||
*/
|
}
|
||||||
public function socketSelect($timeout = 1)
|
|
||||||
{
|
/**
|
||||||
if ($this->count() == 0) {
|
* Returns cURL\Options object with default request's options
|
||||||
throw new Exception('Cannot select if there are no requests in queue.');
|
*
|
||||||
}
|
* @return Options
|
||||||
return curl_multi_select($this->mh, $timeout) !== -1;
|
*/
|
||||||
}
|
public function getDefaultOptions() {
|
||||||
|
if (!isset($this->defaultOptions)) {
|
||||||
|
$this->defaultOptions = new Options();
|
||||||
|
}
|
||||||
|
return $this->defaultOptions;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Overrides default options with given Options object
|
||||||
|
*
|
||||||
|
* @param Options $defaultOptions New options
|
||||||
|
* @return void
|
||||||
|
*/
|
||||||
|
public function setDefaultOptions(Options $defaultOptions) {
|
||||||
|
$this->defaultOptions = $defaultOptions;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Get cURL multi handle
|
||||||
|
*
|
||||||
|
* @return resource
|
||||||
|
*/
|
||||||
|
public function getHandle() {
|
||||||
|
return $this->mh;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Attach request to queue.
|
||||||
|
*
|
||||||
|
* @param Request $request Request to add
|
||||||
|
* @return self
|
||||||
|
*/
|
||||||
|
public function attach(Request $request) {
|
||||||
|
$this->queue[$request->getUID()] = $request;
|
||||||
|
return $this;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Detach request from queue.
|
||||||
|
*
|
||||||
|
* @param Request $request Request to remove
|
||||||
|
* @return self
|
||||||
|
*/
|
||||||
|
public function detach(Request $request) {
|
||||||
|
unset($this->queue[$request->getUID()]);
|
||||||
|
return $this;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Processes handles which are ready and removes them from pool.
|
||||||
|
*
|
||||||
|
* @return int Amount of requests completed
|
||||||
|
*/
|
||||||
|
protected function read() {
|
||||||
|
$n = 0;
|
||||||
|
while ($info = curl_multi_info_read($this->mh)) {
|
||||||
|
$n++;
|
||||||
|
$request = $this->queue[(int) $info['handle']];
|
||||||
|
$result = $info['result'];
|
||||||
|
|
||||||
|
curl_multi_remove_handle($this->mh, $request->getHandle());
|
||||||
|
unset($this->running[$request->getUID()]);
|
||||||
|
$this->detach($request);
|
||||||
|
|
||||||
|
$event = new Event();
|
||||||
|
$event->request = $request;
|
||||||
|
$event->response = new Response($request, curl_multi_getcontent($request->getHandle()));
|
||||||
|
if ($result !== CURLE_OK) {
|
||||||
|
$event->response->setError(new Error(curl_error($request->getHandle()), $result));
|
||||||
|
}
|
||||||
|
$event->queue = $this;
|
||||||
|
$this->dispatch('complete', $event);
|
||||||
|
$request->dispatch('complete', $event);
|
||||||
|
}
|
||||||
|
|
||||||
|
return $n;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Returns count of handles in queue
|
||||||
|
*
|
||||||
|
* @return int Handles count
|
||||||
|
*/
|
||||||
|
public function count() {
|
||||||
|
return count($this->queue);
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Executes requests in parallel
|
||||||
|
*
|
||||||
|
* @return void
|
||||||
|
*/
|
||||||
|
public function send() {
|
||||||
|
while ($this->socketPerform()) {
|
||||||
|
$this->socketSelect();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Returns requests present in $queue but not in $running
|
||||||
|
*
|
||||||
|
* @return Request[] Array of requests
|
||||||
|
*/
|
||||||
|
protected function getRequestsNotRunning() {
|
||||||
|
$map = $this->queue;
|
||||||
|
foreach ($this->running as $k => $v) {
|
||||||
|
unset($map[$k]);
|
||||||
|
}
|
||||||
|
return $map;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Download available data on socket.
|
||||||
|
*
|
||||||
|
* @throws Exception
|
||||||
|
* @return bool TRUE when there are any requests on queue, FALSE when finished
|
||||||
|
*/
|
||||||
|
public function socketPerform() {
|
||||||
|
if ($this->count() == 0) {
|
||||||
|
throw new Exception('Cannot perform if there are no requests in queue.');
|
||||||
|
}
|
||||||
|
|
||||||
|
$notRunning = $this->getRequestsNotRunning();
|
||||||
|
do {
|
||||||
|
/**
|
||||||
|
* Apply cURL options to new requests
|
||||||
|
*/
|
||||||
|
foreach ($notRunning as $request) {
|
||||||
|
$this->getDefaultOptions()->applyTo($request);
|
||||||
|
$request->getOptions()->applyTo($request);
|
||||||
|
curl_multi_add_handle($this->mh, $request->getHandle());
|
||||||
|
$this->running[$request->getUID()] = $request;
|
||||||
|
}
|
||||||
|
|
||||||
|
$runningHandles = null;
|
||||||
|
do {
|
||||||
|
// http://curl.haxx.se/libcurl/c/curl_multi_perform.html
|
||||||
|
// If an added handle fails very quickly, it may never be counted as a running_handle.
|
||||||
|
$mrc = curl_multi_exec($this->mh, $runningHandles);
|
||||||
|
} while ($mrc === CURLM_CALL_MULTI_PERFORM);
|
||||||
|
|
||||||
|
if ($runningHandles < count($this->running)) {
|
||||||
|
$this->read();
|
||||||
|
}
|
||||||
|
|
||||||
|
$notRunning = $this->getRequestsNotRunning();
|
||||||
|
} while (count($notRunning) > 0);
|
||||||
|
// Why the loop? New requests might be added at runtime on 'complete' event.
|
||||||
|
// So we need to attach them to curl_multi handle immediately.
|
||||||
|
|
||||||
|
return $this->count() > 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Waits until activity on socket
|
||||||
|
* On success, returns TRUE. On failure, this function will
|
||||||
|
* return FALSE on a select failure or timeout (from the underlying
|
||||||
|
* select system call)
|
||||||
|
*
|
||||||
|
* @param float|int $timeout Maximum time to wait
|
||||||
|
* @throws Exception
|
||||||
|
* @return bool
|
||||||
|
*/
|
||||||
|
public function socketSelect($timeout = 1) {
|
||||||
|
if ($this->count() == 0) {
|
||||||
|
throw new Exception('Cannot select if there are no requests in queue.');
|
||||||
|
}
|
||||||
|
return curl_multi_select($this->mh, $timeout) !== -1;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@ -49,7 +49,7 @@ class DedimaniaWebHandler {
|
|||||||
|
|
||||||
$asyncHttpRequest = new AsyncHttpRequest($this->maniaControl, self::DEDIMANIA_URL);
|
$asyncHttpRequest = new AsyncHttpRequest($this->maniaControl, self::DEDIMANIA_URL);
|
||||||
$asyncHttpRequest->setCallable(function ($data, $error) use ($updateRecords) {
|
$asyncHttpRequest->setCallable(function ($data, $error) use ($updateRecords) {
|
||||||
Logger::log("Try to connect on Dedimania");
|
Logger::logInfo("Try to connect on Dedimania");
|
||||||
|
|
||||||
if (!$data || $error) {
|
if (!$data || $error) {
|
||||||
Logger::logError("Dedimania Error while opening session: '{$error}' Line 42");
|
Logger::logError("Dedimania Error while opening session: '{$error}' Line 42");
|
||||||
@ -69,7 +69,7 @@ class DedimaniaWebHandler {
|
|||||||
$responseData = $methodResponse[0];
|
$responseData = $methodResponse[0];
|
||||||
$this->dedimaniaData->sessionId = $responseData['SessionId'];
|
$this->dedimaniaData->sessionId = $responseData['SessionId'];
|
||||||
if ($this->dedimaniaData->sessionId) {
|
if ($this->dedimaniaData->sessionId) {
|
||||||
Logger::log("Dedimania connection successfully established.");
|
Logger::logInfo("Dedimania connection successfully established.");
|
||||||
|
|
||||||
if ($updateRecords) {
|
if ($updateRecords) {
|
||||||
$this->fetchDedimaniaRecords();
|
$this->fetchDedimaniaRecords();
|
||||||
@ -146,7 +146,7 @@ class DedimaniaWebHandler {
|
|||||||
$this->dedimaniaData->records[$key] = new RecordData($record);
|
$this->dedimaniaData->records[$key] = new RecordData($record);
|
||||||
}
|
}
|
||||||
|
|
||||||
Logger::log(count($this->dedimaniaData->records) . " Dedimania Records Fetched succesfully!");
|
Logger::logInfo(count($this->dedimaniaData->records) . " Dedimania Records Fetched succesfully!");
|
||||||
|
|
||||||
$this->maniaLinkNeedsUpdate = true;
|
$this->maniaLinkNeedsUpdate = true;
|
||||||
$this->maniaControl->getCallbackManager()->triggerCallback(DedimaniaPlugin::CB_DEDIMANIA_UPDATED, $this->dedimaniaData->records); //TODO
|
$this->maniaControl->getCallbackManager()->triggerCallback(DedimaniaPlugin::CB_DEDIMANIA_UPDATED, $this->dedimaniaData->records); //TODO
|
||||||
@ -233,7 +233,7 @@ class DedimaniaWebHandler {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if (!$record->vReplay) {
|
if (!$record->vReplay) {
|
||||||
Logger::log("Ignore time for " . $record->login . " no validation replay found");
|
Logger::logInfo("Ignore time for " . $record->login . " no validation replay found");
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -256,7 +256,7 @@ class DedimaniaWebHandler {
|
|||||||
$data = array($this->dedimaniaData->sessionId, $this->getMapInfo(), $gameMode, $times, $replays);
|
$data = array($this->dedimaniaData->sessionId, $this->getMapInfo(), $gameMode, $times, $replays);
|
||||||
$content = $this->encodeRequest(self::DEDIMANIA_SET_CHALLENGE_TIMES, $data);
|
$content = $this->encodeRequest(self::DEDIMANIA_SET_CHALLENGE_TIMES, $data);
|
||||||
|
|
||||||
Logger::log("Dedimania Submitting Map Times at End-Map Start");
|
Logger::logInfo("Dedimania Submitting Map Times at End-Map Start");
|
||||||
|
|
||||||
|
|
||||||
$asyncHttpRequest = new AsyncHttpRequest($this->maniaControl, self::DEDIMANIA_URL);
|
$asyncHttpRequest = new AsyncHttpRequest($this->maniaControl, self::DEDIMANIA_URL);
|
||||||
@ -280,7 +280,7 @@ class DedimaniaWebHandler {
|
|||||||
if (!$methodResponse[0]) {
|
if (!$methodResponse[0]) {
|
||||||
Logger::logError("Records Plugin: Submitting dedimania records failed.");
|
Logger::logError("Records Plugin: Submitting dedimania records failed.");
|
||||||
} else {
|
} else {
|
||||||
Logger::log("Dedimania Times succesfully submitted");
|
Logger::logInfo("Dedimania Times succesfully submitted");
|
||||||
}
|
}
|
||||||
|
|
||||||
});
|
});
|
||||||
@ -332,7 +332,7 @@ class DedimaniaWebHandler {
|
|||||||
$this->fetchDedimaniaRecords(true);
|
$this->fetchDedimaniaRecords(true);
|
||||||
}
|
}
|
||||||
|
|
||||||
Logger::log("Dedimania Player added " . $dediPlayer->login);
|
Logger::logInfo("Dedimania Player added " . $dediPlayer->login);
|
||||||
|
|
||||||
$this->maniaLinkNeedsUpdate = true; //TODO handle update for only one player instead of everyone as soon splitted
|
$this->maniaLinkNeedsUpdate = true; //TODO handle update for only one player instead of everyone as soon splitted
|
||||||
});
|
});
|
||||||
@ -376,7 +376,7 @@ class DedimaniaWebHandler {
|
|||||||
$this->handleXmlRpcFault($methodResponse, self::DEDIMANIA_PLAYERDISCONNECT);
|
$this->handleXmlRpcFault($methodResponse, self::DEDIMANIA_PLAYERDISCONNECT);
|
||||||
}
|
}
|
||||||
|
|
||||||
Logger::log("Debug: Dedimania Player left");
|
Logger::logInfo("Debug: Dedimania Player removed");
|
||||||
});
|
});
|
||||||
|
|
||||||
$asyncHttpRequest->setContent($content);
|
$asyncHttpRequest->setContent($content);
|
||||||
@ -420,7 +420,7 @@ class DedimaniaWebHandler {
|
|||||||
$this->handleXmlRpcFault($methodResponse, self::DEDIMANIA_UPDATE_SERVER_PLAYERS);
|
$this->handleXmlRpcFault($methodResponse, self::DEDIMANIA_UPDATE_SERVER_PLAYERS);
|
||||||
}
|
}
|
||||||
|
|
||||||
Logger::log("Dedimania Playerlist Updated");
|
Logger::logInfo("Dedimania Playerlist Updated");
|
||||||
});
|
});
|
||||||
|
|
||||||
$asyncHttpRequest->setContent($content);
|
$asyncHttpRequest->setContent($content);
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user