diff --git a/composer.json b/composer.json index 87438a71..6cc51e89 100644 --- a/composer.json +++ b/composer.json @@ -17,9 +17,12 @@ }, "require-dev": { "phpunit/phpunit": "^8", - "php-coveralls/php-coveralls": "dev-master", + "php-coveralls/php-coveralls": "^2.7", "php-mock/php-mock-phpunit": "^2.6" }, + "suggest": { + "google/cloud-tasks": "Required to use Cloud Tasks push queue routing (^2.3)." + }, "autoload": { "psr-4": { "Google\\AppEngine\\": "src" diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index fdfa520f..31907536 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -50,6 +50,36 @@ final class PushQueue { 'DELETE' => RequestMethod::DELETE ]; + private static $cloudTasksClientFactory = null; + private static $cloudTasksClient = null; + private static $cloudTasksClientClass = null; + + public static function setCloudTasksClientFactory($factory) { + if (self::$cloudTasksClient !== null) { + if (method_exists(self::$cloudTasksClient, 'close')) { + self::$cloudTasksClient->close(); + } + self::$cloudTasksClient = null; + self::$cloudTasksClientClass = null; + } + self::$cloudTasksClientFactory = $factory; + } + + private static function getCloudTasksClient($clientClass) { + if (self::$cloudTasksClient === null || self::$cloudTasksClientClass !== $clientClass) { + if (self::$cloudTasksClient !== null && method_exists(self::$cloudTasksClient, 'close')) { + self::$cloudTasksClient->close(); + } + if (self::$cloudTasksClientFactory !== null) { + self::$cloudTasksClient = call_user_func(self::$cloudTasksClientFactory, $clientClass); + } else { + self::$cloudTasksClient = new $clientClass(); + } + self::$cloudTasksClientClass = $clientClass; + } + return self::$cloudTasksClient; + } + /** * Construct a PushQueue * @@ -142,22 +172,33 @@ public function addTasks($tasks) { if (empty($tasks)) { return []; } + $tasks = array_values($tasks); if (count($tasks) > self::MAX_TASKS_PER_ADD) { throw new \InvalidArgumentException( '$tasks must contain at most ' . self::MAX_TASKS_PER_ADD . ' tasks. Actual size: ' . count($tasks)); } - $req = new TaskQueueBulkAddRequest(); - $resp = new TaskQueueBulkAddResponse(); - - $names = []; - $current_time = microtime(true); foreach ($tasks as $task) { if (!($task instanceof PushTask)) { throw new \InvalidArgumentException( 'All values in $tasks must be instances of PushTask. ' . 'Actual type: ' . gettype($task)); } + } + + $useCloudTasks = + strtolower((string) getenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE')) === 'true' || + getenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE') === '1'; + if ($useCloudTasks) { + return $this->addTasksCloudTasks($tasks); + } + + $req = new TaskQueueBulkAddRequest(); + $resp = new TaskQueueBulkAddResponse(); + + $names = []; + $current_time = microtime(true); + foreach ($tasks as $task) { $names[] = $task->getName(); $add = $req->addAddRequest(); $add->setQueueName($this->name); @@ -212,4 +253,496 @@ public function addTasks($tasks) { } return $names; } + + private static function getMetadataValue($path) { + $opts = [ + 'http' => [ + 'method' => 'GET', + 'header' => 'Metadata-Flavor: Google', + 'timeout' => 1.0 + ] + ]; + $context = stream_context_create($opts); + $url = 'http://metadata.google.internal/computeMetadata/v1/' . $path; + $result = @file_get_contents($url, false, $context); + return $result; + } + + private static function normalizeRegion($region) { + $legacyMap = [ + 'us-central' => 'us-central1', + 'europe-west' => 'europe-west1', + ]; + return isset($legacyMap[$region]) ? $legacyMap[$region] : $region; + } + + private static function getRegion() { + $envRegion = getenv('LOCATION_ID') ?: getenv('GAE_LOCATION') ?: getenv('GAE_REGION') ?: getenv('REGION_ID'); + if ($envRegion) { + return self::normalizeRegion($envRegion); + } + static $region = null; + if ($region === null) { + $regionPath = self::getMetadataValue('instance/region'); + if ($regionPath) { + $parts = explode('/', $regionPath); + $region = end($parts) ?: null; + } + } + if ($region !== null) { + return self::normalizeRegion($region); + } + // The fallback is not cached, so a failed metadata server request (for + // example a timeout during a cold start) is retried on the next call. + return self::normalizeRegion(getenv('LOCAL_GCP_REGION') ?: 'us-central1'); + } + + private static function getProjectId() { + $envProject = getenv('GOOGLE_CLOUD_PROJECT'); + if ($envProject) { + return $envProject; + } + static $projectId = null; + if ($projectId === null) { + $projectId = self::getMetadataValue('project/project-id') ?: null; + } + if ($projectId !== null) { + return $projectId; + } + // Not cached, so a failed metadata server request is retried next time. + $appId = ApiProxy::getCurrentAppId(); + if (($pos = strpos($appId, '~')) !== false) { + return substr($appId, $pos + 1); + } + return $appId; + } + + /** + * Returns the task's headers for a Cloud Tasks AppEngineHttpRequest. + * + * Cloud Tasks sets Host, X-Google-* and X-AppEngine-* headers itself (for + * example X-AppEngine-QueueName and X-AppEngine-TaskName) and ignores + * client supplied values, so they are not sent. + */ + private static function buildCloudTaskHeaders($task) { + $headers = []; + $hasContentType = false; + foreach ($task->getHeaders() as $header) { + $pair = explode(':', $header, 2); + $key = trim($pair[0]); + $val = trim($pair[1]); + if (strcasecmp($key, 'Host') === 0 || + stripos($key, 'X-Google-') === 0 || + stripos($key, 'X-AppEngine-') === 0) { + continue; + } + if (strcasecmp($key, 'Content-Type') === 0) { + $hasContentType = true; + $key = 'Content-Type'; + } + $headers[$key] = $val; + } + if (!$hasContentType) { + $headers['Content-Type'] = 'application/octet-stream'; + } + return $headers; + } + + /** + * Returns the App Engine routing of a task as [service, version, instance], + * with null for parts that are not set. + * + * Like the legacy service, a Host header that is a hostname of the app, + * such as v1.worker.my-app.appspot.com or + * v1-dot-worker-dot-my-app.uc.r.appspot.com, selects + * [[instance.]version.]service. Otherwise the task runs on the current + * service and version, as legacy push tasks without a target do. A queue's + * routing override, set from its queue.yaml target, takes precedence. + */ + private static function buildCloudTaskRouting($task) { + foreach ($task->getHeaders() as $header) { + $pair = explode(':', $header, 2); + if (strcasecmp(trim($pair[0]), 'Host') !== 0) { + continue; + } + $host = preg_replace('/:\d+$/', '', strtolower(trim($pair[1]))); + $suffix = '.appspot.com'; + if (substr($host, -strlen($suffix)) !== $suffix) { + break; + } + $host = substr($host, 0, -strlen($suffix)); + if (substr($host, -2) === '.r') { + // Regional hostname: -dot-..r.appspot.com. + $host = substr($host, 0, -2); + $host = substr($host, 0, (int) strrpos($host, '.')); + } + $projectId = strtolower((string) self::getProjectId()); + if (($colonPos = strpos($projectId, ':')) !== false) { + // Domain-scoped project ID: example.com:my-app -> my-app.example.com + $appHost = substr($projectId, $colonPos + 1) . '.' . substr($projectId, 0, $colonPos); + } else { + $appHost = $projectId; + } + if ($appHost === '') { + break; + } + $normalizedHost = str_replace('-dot-', '.', $host); + if ($normalizedHost === $appHost) { + return ['default', null, null]; + } + $appSuffix = '.' . $appHost; + if (substr($normalizedHost, -strlen($appSuffix)) !== $appSuffix) { + break; + } + $targetPrefix = substr($normalizedHost, 0, -strlen($appSuffix)); + $parts = explode('.', $targetPrefix); + $n = count($parts); + return [$parts[$n - 1], $n > 1 ? $parts[$n - 2] : null, + $n > 2 ? $parts[$n - 3] : null]; + } + return [getenv('GAE_SERVICE') ?: null, getenv('GAE_VERSION') ?: null, null]; + } + + /** + * Returns an AppEngineRouting of $routingClass for the task, or null if no + * routing applies. + */ + private static function newCloudTaskRouting($routingClass, $task) { + list($service, $version, $instance) = self::buildCloudTaskRouting($task); + if ($service === null && $version === null && $instance === null) { + return null; + } + $routing = new $routingClass(); + if ($service !== null) { + $routing->setService($service); + } + if ($version !== null) { + $routing->setVersion($version); + } + if ($instance !== null) { + $routing->setInstance($instance); + } + return $routing; + } + + /** + * Maps a Cloud Tasks ApiException to the TaskQueue exception the legacy + * service would raise. + */ + private static function apiExceptionToTaskQueueException($e, $operation) { + $status = method_exists($e, 'getStatus') ? $e->getStatus() : ''; + $code = $e->getCode(); + $msg = $e->getMessage(); + if ($status === 'ALREADY_EXISTS' || self::isAlreadyExistsError($code, $msg)) { + return new TaskAlreadyExistsException('Task exists already: ' . $msg); + } + if (self::isTransientError($status, $code)) { + return new TransientTaskQueueException( + 'Temporary error, please re-try: ' . $msg); + } + return new TaskQueueException( + 'Cloud Tasks Client SDK ' . $operation . ' failed: ' . $msg); + } + + /** + * Returns true for gRPC statuses that the legacy service reports as + * TRANSIENT_ERROR: DEADLINE_EXCEEDED (4), RESOURCE_EXHAUSTED (8), + * ABORTED (10) and UNAVAILABLE (14). + */ + private static function isTransientError($status, $code) { + return in_array($status, ['DEADLINE_EXCEEDED', 'RESOURCE_EXHAUSTED', + 'ABORTED', 'UNAVAILABLE'], true) || + in_array($code, [4, 8, 10, 14], true); + } + + private function buildCloudTaskObjV2($task, $fullQueueName) { + $headers = self::buildCloudTaskHeaders($task); + $taskName = $task->getName(); + + $methodMap = [ + 'POST' => 1, + 'GET' => 2, + 'HEAD' => 3, + 'PUT' => 4, + 'DELETE' => 5, + 'PATCH' => 6, + 'OPTIONS' => 7, + ]; + $httpMethod = isset($methodMap[$task->getMethod()]) ? $methodMap[$task->getMethod()] : 1; + + $appEngineReq = new \Google\Cloud\Tasks\V2\AppEngineHttpRequest(); + $appEngineReq->setRelativeUri($task->getUrl() ?: '/'); + $appEngineReq->setHttpMethod($httpMethod); + $routing = self::newCloudTaskRouting('\Google\Cloud\Tasks\V2\AppEngineRouting', $task); + if ($routing !== null) { + $appEngineReq->setAppEngineRouting($routing); + } + + foreach ($headers as $k => $v) { + $appEngineReq->getHeaders()[$k] = $v; + } + + if ($task->getMethod() === 'POST' || $task->getMethod() === 'PUT') { + if ($task->getQueryData()) { + $body = http_build_query($task->getQueryData()); + if (strlen($body) > PushTask::MAX_TASK_SIZE_BYTES) { + throw new TaskQueueException('Task greater than maximum size of ' . + PushTask::MAX_TASK_SIZE_BYTES . '. size: ' . strlen($body)); + } + $appEngineReq->setBody($body); + } + } + + $taskObj = new \Google\Cloud\Tasks\V2\Task(); + if ($taskName) { + $fullTaskName = $fullQueueName . "/tasks/" . $taskName; + $taskObj->setName($fullTaskName); + } + $taskObj->setAppEngineHttpRequest($appEngineReq); + + if ($task->getDelaySeconds() > 0) { + $taskObj->setScheduleTime(self::buildScheduleTime($task->getDelaySeconds())); + } + + return $taskObj; + } + + /** + * Builds a Timestamp for now + $delaySeconds, preserving fractional seconds. + */ + private static function buildScheduleTime($delaySeconds) { + $scheduleTime = microtime(true) + $delaySeconds; + $seconds = (int) floor($scheduleTime); + $nanos = (int) round(($scheduleTime - $seconds) * 1e9); + if ($nanos >= 1000000000) { + $seconds += 1; + $nanos -= 1000000000; + } + $ts = new \Google\Protobuf\Timestamp(); + $ts->setSeconds($seconds); + $ts->setNanos($nanos); + return $ts; + } + + /** + * Returns [clientClass, usesRequestObjects] for the Cloud Tasks V2 client + * installed, or null if no Cloud Tasks V2 client is available. + */ + private static function singleTaskClientSpec() { + $candidates = [ + ['\Google\Cloud\Tasks\V2\Client\CloudTasksClient', true], + ['\Google\Cloud\Tasks\V2\CloudTasksClient', false], + ]; + foreach ($candidates as $candidate) { + if (class_exists($candidate[0])) { + return $candidate; + } + } + return null; + } + + /** + * Creates a single task using the shared Cloud Tasks V2 client (or $client + * when provided). + */ + private function createSingleTaskCloudTasks($task, $fullQueueName, $client = null) { + $spec = self::singleTaskClientSpec(); + if ($spec === null) { + throw new TaskQueueException('Cloud Tasks Client SDK is not available.'); + } + list($clientClass, $usesRequestObjects) = $spec; + + $taskObj = $this->buildCloudTaskObjV2($task, $fullQueueName); + $requestClass = '\Google\Cloud\Tasks\V2\CreateTaskRequest'; + + if ($client === null) { + $client = self::getCloudTasksClient($clientClass); + } + try { + if ($usesRequestObjects) { + $createTaskReq = (new $requestClass()) + ->setParent($fullQueueName) + ->setTask($taskObj); + $response = $client->createTask($createTaskReq); + } else { + $response = $client->createTask($fullQueueName, $taskObj); + } + $parts = explode('/', $response->getName()); + return [end($parts)]; + } catch (\Google\ApiCore\ApiException $e) { + throw self::apiExceptionToTaskQueueException($e, 'createTask'); + } + } + + private static function processBatchCreateResponse($response, $tasks, &$names) { + if (method_exists($response, 'isDone') && !$response->isDone()) { + throw new TaskQueueException('Cloud Tasks batch create operation returned done=false'); + } + + $resObj = method_exists($response, 'getResult') ? $response->getResult() : + (method_exists($response, 'getResponse') ? $response->getResponse() : $response); + if ($resObj && method_exists($resObj, 'getTasks')) { + foreach ($resObj->getTasks() as $resTask) { + $parts = explode('/', $resTask->getName()); + $names[] = end($parts); + } + } + + $metadata = method_exists($response, 'getMetadata') ? $response->getMetadata() : null; + $failedRequests = ($metadata && method_exists($metadata, 'getFailedRequests')) ? $metadata->getFailedRequests() : null; + $exception = null; + foreach ($tasks as $idx => $task) { + $hasFailure = $failedRequests && + ((is_array($failedRequests) && isset($failedRequests[$idx])) || + (is_object($failedRequests) && method_exists($failedRequests, 'offsetExists') && $failedRequests->offsetExists($idx))); + if (!$hasFailure) { + continue; + } + $errStatus = is_array($failedRequests) ? $failedRequests[$idx] : $failedRequests->offsetGet($idx); + $code = ($errStatus && method_exists($errStatus, 'getCode')) ? $errStatus->getCode() : 0; + if ($code !== 0) { + $msg = ($errStatus && method_exists($errStatus, 'getMessage')) ? $errStatus->getMessage() : ''; + if (self::isAlreadyExistsError($code, $msg)) { + $exception = new TaskAlreadyExistsException('Task exists already: ' . $msg); + } elseif (stripos($msg, 'queue does not exist') !== false || stripos($msg, 'queue no longer exists') !== false) { + throw new TaskQueueException('Unknown queue: ' . $msg); + } elseif (self::isTransientError('', $code)) { + throw new TransientTaskQueueException('Temporary error, please re-try: ' . $msg); + } else { + throw new TaskQueueException('Task creation failed: ' . $msg); + } + } + } + if ($exception !== null) { + throw $exception; + } + } + + /** + * Returns true if $client->$methodName() accepts a Request object rather + * than positional arguments ($parent, ...). + */ + private static function methodAcceptsRequestObject($client, $methodName) { + $ref = new \ReflectionMethod($client, $methodName); + $params = $ref->getParameters(); + if (empty($params)) { + return true; + } + $firstParam = $params[0]; + if (method_exists($firstParam, 'getType') && $firstParam->getType() !== null) { + $type = $firstParam->getType(); + $typeName = method_exists($type, 'getName') ? $type->getName() : (string) $type; + return $typeName !== 'string'; + } + $paramName = strtolower($firstParam->getName()); + if ($paramName === 'parent' || $paramName === 'formattedparent') { + return false; + } + return $ref->getNumberOfRequiredParameters() < 2; + } + + private function addTasksCloudTasks($tasks) { + $tasks = array_values($tasks); + $projectId = self::getProjectId(); + $region = self::getRegion(); + $fullQueueName = "projects/" . $projectId . "/locations/" . $region . "/queues/" . $this->name; + + if (count($tasks) === 1) { + return $this->createSingleTaskCloudTasks($tasks[0], $fullQueueName); + } + + // addTasks() allows at most MAX_TASKS_PER_ADD (100) tasks, which is also + // the BatchCreateTasks limit, so every call is a single batch request. + $names = []; + + $v2ClientClass = null; + if (class_exists('\Google\Cloud\Tasks\V2\Client\CloudTasksClient') && method_exists('\Google\Cloud\Tasks\V2\Client\CloudTasksClient', 'batchCreateTasks')) { + $v2ClientClass = '\Google\Cloud\Tasks\V2\Client\CloudTasksClient'; + } elseif (class_exists('\Google\Cloud\Tasks\V2\CloudTasksClient') && method_exists('\Google\Cloud\Tasks\V2\CloudTasksClient', 'batchCreateTasks')) { + $v2ClientClass = '\Google\Cloud\Tasks\V2\CloudTasksClient'; + } + + if ($v2ClientClass !== null) { + $client = self::getCloudTasksClient($v2ClientClass); + $createTaskRequests = []; + foreach ($tasks as $task) { + $taskObj = $this->buildCloudTaskObjV2($task, $fullQueueName); + $createTaskReq = (new \Google\Cloud\Tasks\V2\CreateTaskRequest()) + ->setParent($fullQueueName) + ->setTask($taskObj); + $createTaskRequests[] = $createTaskReq; + } + + try { + if (class_exists('\Google\Cloud\Tasks\V2\BatchCreateTasksRequest') && + self::methodAcceptsRequestObject($client, 'batchCreateTasks')) { + $batchReq = (new \Google\Cloud\Tasks\V2\BatchCreateTasksRequest()) + ->setParent($fullQueueName) + ->setRequests($createTaskRequests); + $response = $client->batchCreateTasks($batchReq); + } else { + $response = $client->batchCreateTasks($fullQueueName, $createTaskRequests); + } + + self::processBatchCreateResponse($response, $tasks, $names); + } catch (\Google\ApiCore\ApiException $e) { + throw self::apiExceptionToTaskQueueException($e, 'batchCreate'); + } + return $names; + } + + // Fallback: If native batchCreateTasks is not available in the installed Cloud Tasks V2 SDK, + // enqueue tasks individually using createTask over the shared client. + $spec = self::singleTaskClientSpec(); + if ($spec === null) { + throw new TaskQueueException('Cloud Tasks Client SDK is not available.'); + } + $client = self::getCloudTasksClient($spec[0]); + return $this->addTasksIndividually($tasks, $fullQueueName, $client); + } + + /** + * Creates tasks one at a time. Like BulkAdd, every task is attempted, and + * TaskAlreadyExistsException is only thrown if no task failed for another + * reason, so that it still means every task has been added. + */ + private function addTasksIndividually($tasks, $fullQueueName, $client) { + syslog( + LOG_WARNING, + 'Cloud Tasks batchCreateTasks is not available in the installed ' . + 'google/cloud-tasks package; falling back to sequential createTask ' . + 'calls. Upgrade google/cloud-tasks to avoid request timeouts on ' . + 'large batches.' + ); + $names = []; + $alreadyExists = null; + $error = null; + foreach ($tasks as $task) { + try { + $res = $this->createSingleTaskCloudTasks($task, $fullQueueName, $client); + $names[] = $res[0]; + } catch (TaskAlreadyExistsException $e) { + $alreadyExists = $alreadyExists ?: $e; + } catch (TaskQueueException $e) { + $error = $error ?: $e; + } + } + if ($error !== null) { + throw $error; + } + if ($alreadyExists !== null) { + throw $alreadyExists; + } + return $names; + } + + private static function isAlreadyExistsError($errCode, $errMsg) { + // Cloud Tasks reports duplicate and tombstoned task names as ALREADY_EXISTS. + // NOT_FOUND is intentionally not mapped here so that missing queues, + // projects or locations surface as TaskQueueException. + return $errCode === 6 || $errCode === 409 || + stripos($errMsg, 'already exists') !== false || + stripos($errMsg, 'existed too recently') !== false || + stripos($errMsg, 'tombstoned') !== false; + } } diff --git a/tests/Api/TaskQueue/PushQueueTest.php b/tests/Api/TaskQueue/PushQueueTest.php index 2901d6ab..7863ba01 100644 --- a/tests/Api/TaskQueue/PushQueueTest.php +++ b/tests/Api/TaskQueue/PushQueueTest.php @@ -30,6 +30,17 @@ class PushQueueTest extends ApiProxyTestBase { + // Used to mock file_get_contents() for metadata server requests. + // See https://github.com/php-mock/php-mock-phpunit + use \phpmock\phpunit\PHPMock; + + public static function setUpBeforeClass(): void { + parent::setUpBeforeClass(); + // Defined up front so the mock works even if file_get_contents() was + // already called in this namespace. + self::defineFunctionMock(__NAMESPACE__, 'file_get_contents'); + } + public function setUp(): void { parent::setUp(); $this->_SERVER = $_SERVER; @@ -144,6 +155,28 @@ public function testPushQueueSimplestAddTasks() { $this->apiProxyMock->verify(); } + public function testGaePushQueueBackendEnvDoesNotEnableCloudTasks() { + putenv('GAE_PUSHQUEUE_BACKEND=CLOUD_TASK'); + try { + PushQueue::setCloudTasksClientFactory(function ($className) { + throw new \LogicException('Cloud Tasks must not be used'); + }); + $req = self::buildBulkAddRequest(); + $resp = new TaskQueueBulkAddResponse(); + $task_result = $resp->addTaskResult(); + $task_result->setResult(ErrorCode::OK); + $task_result->setChosenTaskName('fred'); + $this->apiProxyMock->expectCall('taskqueue', 'BulkAdd', $req, $resp); + + $task_names = (new PushQueue())->addTasks([new PushTask('/someUrl')]); + $this->assertEquals(['fred'], $task_names); + $this->apiProxyMock->verify(); + } finally { + PushQueue::setCloudTasksClientFactory(null); + putenv('GAE_PUSHQUEUE_BACKEND'); + } + } + public function testPushQueueAddTwoTasks() { $req = self::buildBulkAddRequestWithTwoTasks(); @@ -278,4 +311,787 @@ public function testPushQueueTooManyTasksError() { $this->apiProxyMock->verify(); } + private static function ensureCloudTasksV2StubsLoaded() { + if (!class_exists('\Google\Protobuf\Timestamp')) { + eval(' + namespace Google\Protobuf; + class Timestamp { + private $seconds = 0; + private $nanos = 0; + public function setSeconds($s) { $this->seconds = $s; return $this; } + public function getSeconds() { return $this->seconds; } + public function setNanos($n) { $this->nanos = $n; return $this; } + public function getNanos() { return $this->nanos; } + } + '); + } + if (!class_exists('\Google\Rpc\Status')) { + eval(' + namespace Google\Rpc; + class Status { + private $code = 0; + private $message = ""; + public function setCode($c) { $this->code = $c; return $this; } + public function getCode() { return $this->code; } + public function setMessage($m) { $this->message = $m; return $this; } + public function getMessage() { return $this->message; } + } + '); + } + if (!class_exists('\Google\ApiCore\ApiException')) { + eval(' + namespace Google\ApiCore; + class ApiException extends \Exception { + private $status; + public function __construct($message, $code, $status = "") { + parent::__construct($message, $code); + $this->status = $status; + } + public function getStatus() { return $this->status; } + } + '); + } + if (!class_exists('\Google\Cloud\Tasks\V2\Task')) { + eval(' + namespace Google\Cloud\Tasks\V2; + class AppEngineHttpRequest { + private $relativeUri = "/"; + private $httpMethod = 1; + private $headers; + private $body = ""; + private $appEngineRouting = null; + public function __construct() { $this->headers = new \ArrayObject(); } + public function setAppEngineRouting($r) { $this->appEngineRouting = $r; return $this; } + public function getAppEngineRouting() { return $this->appEngineRouting; } + public function setRelativeUri($u) { $this->relativeUri = $u; return $this; } + public function getRelativeUri() { return $this->relativeUri; } + public function setHttpMethod($m) { $this->httpMethod = $m; return $this; } + public function getHttpMethod() { return $this->httpMethod; } + public function getHeaders() { return $this->headers; } + public function setBody($b) { $this->body = $b; return $this; } + public function getBody() { return $this->body; } + } + class AppEngineRouting { + private $service = ""; + private $version = ""; + private $instance = ""; + public function setService($s) { $this->service = $s; return $this; } + public function getService() { return $this->service; } + public function setVersion($v) { $this->version = $v; return $this; } + public function getVersion() { return $this->version; } + public function setInstance($i) { $this->instance = $i; return $this; } + public function getInstance() { return $this->instance; } + } + class Task { + private $name = ""; + private $appEngineHttpRequest = null; + private $scheduleTime = null; + public function setName($n) { $this->name = $n; return $this; } + public function getName() { return $this->name; } + public function setAppEngineHttpRequest($r) { $this->appEngineHttpRequest = $r; return $this; } + public function getAppEngineHttpRequest() { return $this->appEngineHttpRequest; } + public function setScheduleTime($t) { $this->scheduleTime = $t; return $this; } + public function getScheduleTime() { return $this->scheduleTime; } + } + class CreateTaskRequest { + private $parent = ""; + private $task = null; + public function setParent($p) { $this->parent = $p; return $this; } + public function getParent() { return $this->parent; } + public function setTask($t) { $this->task = $t; return $this; } + public function getTask() { return $this->task; } + } + class BatchCreateTasksRequest { + private $parent = ""; + private $requests = []; + public function setParent($p) { $this->parent = $p; return $this; } + public function getParent() { return $this->parent; } + public function setRequests($r) { $this->requests = $r; return $this; } + public function getRequests() { return $this->requests; } + } + class BatchCreateTasksResponse { + private $tasks = []; + public function setTasks($t) { $this->tasks = $t; return $this; } + public function getTasks() { return $this->tasks; } + } + class BatchCreateTasksMetadata { + private $failedRequests = []; + public function setFailedRequests($f) { $this->failedRequests = $f; return $this; } + public function getFailedRequests() { return $this->failedRequests; } + } + '); + } + if (!class_exists('\Google\Cloud\Tasks\V2\Client\CloudTasksClient')) { + eval(' + namespace Google\Cloud\Tasks\V2\Client; + class CloudTasksClient { + public function createTask($req) {} + public function batchCreateTasks($req) {} + public function close() {} + } + '); + } + } + + public function testCloudTasksV2SingleCreateTask() { + self::ensureCloudTasksV2StubsLoaded(); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE=true'); + putenv('GOOGLE_CLOUD_PROJECT=test-proj'); + putenv('LOCATION_ID=us-central1'); + try { + $capturedReq = null; + PushQueue::setCloudTasksClientFactory(function ($className) use (&$capturedReq) { + return new class($capturedReq) { + private $capturedReqRef; + public function __construct(&$ref) { $this->capturedReqRef = &$ref; } + public function createTask($req) { + $this->capturedReqRef = $req; + return (new \Google\Cloud\Tasks\V2\Task()) + ->setName('projects/test-proj/locations/us-central1/queues/default/tasks/single-1'); + } + public function close() {} + }; + }); + + $task = new PushTask('/worker/push', ['foo' => 'bar'], [ + 'name' => 'single-1', + 'delay_seconds' => 30, + ]); + $queue = new PushQueue('default'); + $names = $queue->addTasks([$task]); + $this->assertEquals(['single-1'], $names); + $this->assertNotNull($capturedReq); + $this->assertEquals( + 'projects/test-proj/locations/us-central1/queues/default', + $capturedReq->getParent()); + $ctTask = $capturedReq->getTask(); + $this->assertEquals( + 'projects/test-proj/locations/us-central1/queues/default/tasks/single-1', + $ctTask->getName()); + $this->assertEquals('/worker/push', $ctTask->getAppEngineHttpRequest()->getRelativeUri()); + $this->assertEquals('foo=bar', $ctTask->getAppEngineHttpRequest()->getBody()); + $this->assertNotNull($ctTask->getScheduleTime()); + } finally { + PushQueue::setCloudTasksClientFactory(null); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE'); + putenv('GOOGLE_CLOUD_PROJECT'); + putenv('LOCATION_ID'); + } + } + + public function testCloudTasksV2SingleCreateTaskAlreadyExists() { + self::ensureCloudTasksV2StubsLoaded(); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE=true'); + putenv('GOOGLE_CLOUD_PROJECT=test-proj'); + putenv('LOCATION_ID=us-central1'); + try { + PushQueue::setCloudTasksClientFactory(function ($className) { + return new class { + public function createTask($req) { + throw new \Google\ApiCore\ApiException('Task already exists', 409, 'ALREADY_EXISTS'); + } + public function close() {} + }; + }); + + $this->expectException('\Google\AppEngine\Api\TaskQueue\TaskAlreadyExistsException'); + $task = new PushTask('/worker/push', [], ['name' => 'dup-single']); + $queue = new PushQueue('default'); + $queue->addTasks([$task]); + } finally { + PushQueue::setCloudTasksClientFactory(null); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE'); + putenv('GOOGLE_CLOUD_PROJECT'); + putenv('LOCATION_ID'); + } + } + + public function testCloudTasksV2BatchCreateTasksViaGetResult() { + self::ensureCloudTasksV2StubsLoaded(); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE=true'); + putenv('GOOGLE_CLOUD_PROJECT=test-proj'); + putenv('LOCATION_ID=us-central1'); + try { + PushQueue::setCloudTasksClientFactory(function ($className) { + return new class { + public function batchCreateTasks($req) { + return new class { + public function isDone() { return true; } + public function operationFailed() { return false; } + public function getMetadata() { + return new \Google\Cloud\Tasks\V2\BatchCreateTasksMetadata(); + } + public function getResult() { + $t1 = (new \Google\Cloud\Tasks\V2\Task()) + ->setName('projects/test-proj/locations/us-central1/queues/default/tasks/task-1'); + $t2 = (new \Google\Cloud\Tasks\V2\Task()) + ->setName('projects/test-proj/locations/us-central1/queues/default/tasks/task-2'); + return (new \Google\Cloud\Tasks\V2\BatchCreateTasksResponse()) + ->setTasks([$t1, $t2]); + } + }; + } + public function close() {} + }; + }); + + $task1 = new PushTask('/url1', ['k' => 'v1'], ['name' => 'task-1']); + $task2 = new PushTask('/url2', ['k' => 'v2'], ['name' => 'task-2']); + $queue = new PushQueue('default'); + $names = $queue->addTasks([$task1, $task2]); + $this->assertEquals(['task-1', 'task-2'], $names); + } finally { + PushQueue::setCloudTasksClientFactory(null); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE'); + putenv('GOOGLE_CLOUD_PROJECT'); + putenv('LOCATION_ID'); + } + } + + public function testCloudTasksV2BatchCreateTasksFailedRequestsAlreadyExists() { + self::ensureCloudTasksV2StubsLoaded(); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE=true'); + putenv('GOOGLE_CLOUD_PROJECT=test-proj'); + putenv('LOCATION_ID=us-central1'); + try { + PushQueue::setCloudTasksClientFactory(function ($className) { + return new class { + public function batchCreateTasks($req) { + return new class { + public function isDone() { return true; } + public function operationFailed() { return false; } + public function getMetadata() { + $st = (new \Google\Rpc\Status()) + ->setCode(6) + ->setMessage('The task cannot be created because a task with this name existed too recently'); + return (new \Google\Cloud\Tasks\V2\BatchCreateTasksMetadata()) + ->setFailedRequests([1 => $st]); + } + public function getResult() { + $t1 = (new \Google\Cloud\Tasks\V2\Task()) + ->setName('projects/test-proj/locations/us-central1/queues/default/tasks/task-1'); + return (new \Google\Cloud\Tasks\V2\BatchCreateTasksResponse()) + ->setTasks([$t1]); + } + }; + } + public function close() {} + }; + }); + + $this->expectException('\Google\AppEngine\Api\TaskQueue\TaskAlreadyExistsException'); + $task1 = new PushTask('/url1'); + $task2 = new PushTask('/url2', [], ['name' => 'dup-task']); + $queue = new PushQueue('default'); + $queue->addTasks([$task1, $task2]); + } finally { + PushQueue::setCloudTasksClientFactory(null); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE'); + putenv('GOOGLE_CLOUD_PROJECT'); + putenv('LOCATION_ID'); + } + } + + public function testCloudTasksV2BatchCreateTasksFailedRequestsUnknownQueue() { + self::ensureCloudTasksV2StubsLoaded(); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE=true'); + putenv('GOOGLE_CLOUD_PROJECT=test-proj'); + putenv('LOCATION_ID=us-central1'); + try { + PushQueue::setCloudTasksClientFactory(function ($className) { + return new class { + public function batchCreateTasks($req) { + return new class { + public function isDone() { return true; } + public function operationFailed() { return false; } + public function getMetadata() { + $st = (new \Google\Rpc\Status()) + ->setCode(5) + ->setMessage('Queue does not exist'); + return (new \Google\Cloud\Tasks\V2\BatchCreateTasksMetadata()) + ->setFailedRequests([0 => $st]); + } + public function getResult() { + return new \Google\Cloud\Tasks\V2\BatchCreateTasksResponse(); + } + }; + } + public function close() {} + }; + }); + + $this->expectException('\Google\AppEngine\Api\TaskQueue\TaskQueueException'); + $this->expectExceptionMessage('Unknown queue'); + $task1 = new PushTask('/url1'); + $task2 = new PushTask('/url2'); + $queue = new PushQueue('missing-queue'); + $queue->addTasks([$task1, $task2]); + } finally { + PushQueue::setCloudTasksClientFactory(null); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE'); + putenv('GOOGLE_CLOUD_PROJECT'); + putenv('LOCATION_ID'); + } + } + + public function testCloudTasksRoutingFromHostHeader() { + self::ensureCloudTasksV2StubsLoaded(); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE=true'); + putenv('GOOGLE_CLOUD_PROJECT=test-proj'); + putenv('LOCATION_ID=us-central1'); + putenv('GAE_SERVICE=current'); + putenv('GAE_VERSION=cv'); + try { + $captured = []; + PushQueue::setCloudTasksClientFactory(function ($className) use (&$captured) { + return new class($captured) { + private $capturedRef; + public function __construct(&$ref) { $this->capturedRef = &$ref; } + public function createTask($req) { + $this->capturedRef[] = $req; + return (new \Google\Cloud\Tasks\V2\Task()) + ->setName('projects/test-proj/locations/us-central1/queues/default/tasks/t'); + } + public function close() {} + }; + }); + + $cases = [ + // [Host header, service, version, instance] + ['Host: 1.v2.worker.test-proj.appspot.com', 'worker', 'v2', '1'], + ['host: v2-dot-worker-dot-test-proj.uc.r.appspot.com:443', 'worker', 'v2', ''], + ['Host: worker-dot-test-proj.appspot.com', 'worker', '', ''], + ['Host: test-proj.appspot.com', 'default', '', ''], + // Different project ID or not a hostname of the app, and no Host + // header: the current service and version. + ['Host: worker.other-proj.appspot.com', 'current', 'cv', ''], + ['Host: other-proj.appspot.com', 'current', 'cv', ''], + ['Host: example.com', 'current', 'cv', ''], + ['', 'current', 'cv', ''], + ]; + $queue = new PushQueue('default'); + foreach ($cases as $case) { + $queue->addTasks([new PushTask('/w', [], ['header' => $case[0]])]); + $req = end($captured)->getTask()->getAppEngineHttpRequest(); + $routing = $req->getAppEngineRouting(); + $this->assertNotNull($routing, $case[0]); + $this->assertEquals( + [$case[1], $case[2], $case[3]], + [$routing->getService(), $routing->getVersion(), $routing->getInstance()], + $case[0]); + $this->assertFalse(isset($req->getHeaders()['Host']), $case[0]); + $this->assertFalse(isset($req->getHeaders()['host']), $case[0]); + } + + // Domain-scoped project ID: example.com:my-app -> my-app.example.com.appspot.com + putenv('GOOGLE_CLOUD_PROJECT=example.com:my-app'); + $domainCases = [ + ['Host: my-app.example.com.appspot.com', 'default', '', ''], + ['Host: worker.my-app.example.com.appspot.com', 'worker', '', ''], + ['Host: v2-dot-worker-dot-my-app-dot-example.com.uc.r.appspot.com', 'worker', 'v2', ''], + ]; + foreach ($domainCases as $case) { + $queue->addTasks([new PushTask('/w', [], ['header' => $case[0]])]); + $req = end($captured)->getTask()->getAppEngineHttpRequest(); + $routing = $req->getAppEngineRouting(); + $this->assertNotNull($routing, $case[0]); + $this->assertEquals( + [$case[1], $case[2], $case[3]], + [$routing->getService(), $routing->getVersion(), $routing->getInstance()], + $case[0]); + } + } finally { + PushQueue::setCloudTasksClientFactory(null); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE'); + putenv('GOOGLE_CLOUD_PROJECT'); + putenv('LOCATION_ID'); + putenv('GAE_SERVICE'); + putenv('GAE_VERSION'); + } + } + + public function testCloudTasksV2SingleCreateTaskFractionalDelay() { + self::ensureCloudTasksV2StubsLoaded(); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE=true'); + putenv('GOOGLE_CLOUD_PROJECT=test-proj'); + putenv('LOCATION_ID=us-central1'); + try { + $capturedReq = null; + PushQueue::setCloudTasksClientFactory(function ($className) use (&$capturedReq) { + return new class($capturedReq) { + private $capturedReqRef; + public function __construct(&$ref) { $this->capturedReqRef = &$ref; } + public function createTask($req) { + $this->capturedReqRef = $req; + return (new \Google\Cloud\Tasks\V2\Task()) + ->setName('projects/test-proj/locations/us-central1/queues/default/tasks/frac-1'); + } + public function close() {} + }; + }); + + MockMicrotime::expect(12345.25); + $task = new PushTask('/worker/push', [], [ + 'name' => 'frac-1', + 'delay_seconds' => 0.5, + ]); + $queue = new PushQueue('default'); + $queue->addTasks([$task]); + + $scheduleTime = $capturedReq->getTask()->getScheduleTime(); + $this->assertEquals(12345, $scheduleTime->getSeconds()); + $this->assertEquals(750000000, $scheduleTime->getNanos()); + } finally { + PushQueue::setCloudTasksClientFactory(null); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE'); + putenv('GOOGLE_CLOUD_PROJECT'); + putenv('LOCATION_ID'); + } + } + + public function testCloudTasksV2SingleCreateTaskNotFoundIsNotAlreadyExists() { + self::ensureCloudTasksV2StubsLoaded(); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE=true'); + putenv('GOOGLE_CLOUD_PROJECT=test-proj'); + putenv('LOCATION_ID=us-central1'); + try { + PushQueue::setCloudTasksClientFactory(function ($className) { + return new class { + public function createTask($req) { + throw new \Google\ApiCore\ApiException('Requested entity was not found.', 5, 'NOT_FOUND'); + } + public function close() {} + }; + }); + + $task = new PushTask('/worker/push', [], ['name' => 'nf-1']); + $queue = new PushQueue('default'); + try { + $queue->addTasks([$task]); + $this->fail('Expected TaskQueueException'); + } catch (TaskQueueException $e) { + $this->assertNotInstanceOf(TaskAlreadyExistsException::class, $e); + $this->assertStringContainsString('Requested entity was not found', $e->getMessage()); + } + } finally { + PushQueue::setCloudTasksClientFactory(null); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE'); + putenv('GOOGLE_CLOUD_PROJECT'); + putenv('LOCATION_ID'); + } + } + + public function testCloudTasksV2BatchCreateTasksUsesSingleClient() { + self::ensureCloudTasksV2StubsLoaded(); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE=true'); + putenv('GOOGLE_CLOUD_PROJECT=test-proj'); + putenv('LOCATION_ID=us-central1'); + try { + $counts = ['created' => 0, 'closed' => 0]; + PushQueue::setCloudTasksClientFactory(function ($className) use (&$counts) { + $counts['created']++; + return new class($counts) { + private $countsRef; + public function __construct(&$ref) { $this->countsRef = &$ref; } + public function batchCreateTasks($req) { + return new class { + public function isDone() { return true; } + public function operationFailed() { return false; } + public function getMetadata() { + return new \Google\Cloud\Tasks\V2\BatchCreateTasksMetadata(); + } + public function getResult() { + $t1 = (new \Google\Cloud\Tasks\V2\Task()) + ->setName('projects/test-proj/locations/us-central1/queues/default/tasks/a'); + $t2 = (new \Google\Cloud\Tasks\V2\Task()) + ->setName('projects/test-proj/locations/us-central1/queues/default/tasks/b'); + return (new \Google\Cloud\Tasks\V2\BatchCreateTasksResponse()) + ->setTasks([$t1, $t2]); + } + }; + } + public function close() { $this->countsRef['closed']++; } + }; + }); + + $queue = new PushQueue('default'); + $names = $queue->addTasks([ + new PushTask('/a', [], ['name' => 'a']), + new PushTask('/b', [], ['name' => 'b']), + ]); + $this->assertEquals(['a', 'b'], $names); + $this->assertEquals(1, $counts['created']); + $this->assertEquals(0, $counts['closed']); + // A second batch call reuses the same cached client. + $queue->addTasks([ + new PushTask('/a', [], ['name' => 'a']), + new PushTask('/b', [], ['name' => 'b']), + ]); + $this->assertEquals(1, $counts['created']); + PushQueue::setCloudTasksClientFactory(null); + $this->assertEquals(1, $counts['closed']); + } finally { + PushQueue::setCloudTasksClientFactory(null); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE'); + putenv('GOOGLE_CLOUD_PROJECT'); + putenv('LOCATION_ID'); + } + } + + public function testCloudTasksV2NonZeroIndexedTasksArray() { + self::ensureCloudTasksV2StubsLoaded(); + self::setCloudTasksEnv(); + try { + PushQueue::setCloudTasksClientFactory(function ($className) { + return new class { + public function createTask($req) { + return (new \Google\Cloud\Tasks\V2\Task()) + ->setName($req->getTask()->getName() ?: 'projects/test-proj/locations/us-central1/queues/default/tasks/single-nz'); + } + public function batchCreateTasks($req) { + return new class { + public function getMetadata() { + $st = (new \Google\Rpc\Status()) + ->setCode(6) + ->setMessage('Task already exists'); + return (new \Google\Cloud\Tasks\V2\BatchCreateTasksMetadata()) + ->setFailedRequests([1 => $st]); + } + public function getResult() { + $t1 = (new \Google\Cloud\Tasks\V2\Task()) + ->setName('projects/test-proj/locations/us-central1/queues/default/tasks/t1'); + return (new \Google\Cloud\Tasks\V2\BatchCreateTasksResponse()) + ->setTasks([$t1]); + } + }; + } + public function close() {} + }; + }); + + $queue = new PushQueue('default'); + $singleRes = $queue->addTasks([2 => new PushTask('/url1', [], ['name' => 'single-nz'])]); + $this->assertEquals(['single-nz'], $singleRes); + + $this->expectException(TaskAlreadyExistsException::class); + $queue->addTasks([ + 5 => new PushTask('/url1', [], ['name' => 't1']), + 9 => new PushTask('/url2', [], ['name' => 't2']), + ]); + } finally { + self::clearCloudTasksEnv(); + } + } + + private static function setCloudTasksEnv() { + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE=true'); + putenv('GOOGLE_CLOUD_PROJECT=test-proj'); + putenv('LOCATION_ID=us-central1'); + } + + private static function clearCloudTasksEnv() { + PushQueue::setCloudTasksClientFactory(null); + putenv('APPENGINE_USE_CLOUDTASK_PUSH_QUEUE'); + putenv('GOOGLE_CLOUD_PROJECT'); + putenv('LOCATION_ID'); + } + + public function testCloudTasksV2DoesNotSendReservedHeaders() { + self::ensureCloudTasksV2StubsLoaded(); + self::setCloudTasksEnv(); + try { + $capturedReq = null; + PushQueue::setCloudTasksClientFactory(function ($className) use (&$capturedReq) { + return new class($capturedReq) { + private $capturedReqRef; + public function __construct(&$ref) { $this->capturedReqRef = &$ref; } + public function createTask($req) { + $this->capturedReqRef = $req; + return (new \Google\Cloud\Tasks\V2\Task()) + ->setName('projects/test-proj/locations/us-central1/queues/default/tasks/h-1'); + } + public function close() {} + }; + }); + + $task = new PushTask('/worker', ['a' => 'b'], [ + 'name' => 'h-1', + 'header' => "X-AppEngine-QueueName: spoofed\r\nX-Google-Foo: x\r\n" . + "Host: example.com\r\nX-Custom: ok", + ]); + (new PushQueue('default'))->addTasks([$task]); + + $headers = $capturedReq->getTask()->getAppEngineHttpRequest()->getHeaders(); + $this->assertEquals('ok', $headers['X-Custom']); + $this->assertEquals('application/x-www-form-urlencoded', $headers['Content-Type']); + foreach (['X-AppEngine-QueueName', 'X-AppEngine-TaskName', 'X-Google-Foo', 'Host'] as $reserved) { + $this->assertFalse(isset($headers[$reserved]), $reserved . ' should not be sent'); + } + } finally { + self::clearCloudTasksEnv(); + } + } + + public function testRegionMetadataFallbackIsNotCached() { + $saved = []; + foreach (['LOCATION_ID', 'GAE_LOCATION', 'GAE_REGION', 'REGION_ID', 'LOCAL_GCP_REGION'] as $var) { + $saved[$var] = getenv($var); + putenv($var); + } + try { + $fileGetContents = $this->getFunctionMock(__NAMESPACE__, 'file_get_contents'); + $fileGetContents->expects($this->exactly(2))->willReturnOnConsecutiveCalls( + false, 'projects/123/regions/europe-west1'); + $getRegion = new \ReflectionMethod(PushQueue::class, 'getRegion'); + $getRegion->setAccessible(true); + + // A failed metadata request falls back without caching the fallback... + $this->assertEquals('us-central1', $getRegion->invoke(null)); + // ...so the next call asks the metadata server again and caches it. + $this->assertEquals('europe-west1', $getRegion->invoke(null)); + $this->assertEquals('europe-west1', $getRegion->invoke(null)); + } finally { + foreach ($saved as $var => $value) { + putenv($value === false ? $var : "$var=$value"); + } + } + } + + public function testCloudTasksV2TransientErrorIsTransientException() { + self::ensureCloudTasksV2StubsLoaded(); + self::setCloudTasksEnv(); + try { + PushQueue::setCloudTasksClientFactory(function ($className) { + return new class { + public function createTask($req) { + throw new \Google\ApiCore\ApiException('The service is unavailable.', 14, 'UNAVAILABLE'); + } + public function close() {} + }; + }); + + $this->expectException(TransientTaskQueueException::class); + (new PushQueue('default'))->addTasks([new PushTask('/worker')]); + } finally { + self::clearCloudTasksEnv(); + } + } + + public function testAddTasksIndividuallyAttemptsEveryTask() { + self::ensureCloudTasksV2StubsLoaded(); + $addTasksIndividually = new \ReflectionMethod(PushQueue::class, 'addTasksIndividually'); + $addTasksIndividually->setAccessible(true); + $newClient = function ($errors) { + return new class($errors) { + public $calls = 0; + private $errors; + public function __construct($errors) { $this->errors = $errors; } + public function createTask($req) { + $error = $this->errors[$this->calls++]; + if ($error !== null) { + throw $error; + } + return (new \Google\Cloud\Tasks\V2\Task())->setName($req->getTask()->getName()); + } + public function close() {} + }; + }; + $queueName = 'projects/test-proj/locations/us-central1/queues/default'; + $tasks = [ + new PushTask('/a', [], ['name' => 'a']), + new PushTask('/b', [], ['name' => 'b']), + new PushTask('/c', [], ['name' => 'c']), + ]; + $alreadyExists = new \Google\ApiCore\ApiException('already exists', 6, 'ALREADY_EXISTS'); + $notFound = new \Google\ApiCore\ApiException('Queue does not exist.', 5, 'NOT_FOUND'); + + // A duplicate does not stop the remaining tasks from being added. + $client = $newClient([$alreadyExists, null, null]); + try { + $addTasksIndividually->invoke(new PushQueue('default'), $tasks, $queueName, $client); + $this->fail('Expected TaskAlreadyExistsException'); + } catch (TaskAlreadyExistsException $e) { + $this->assertEquals(3, $client->calls); + } + + // Another error takes precedence over TaskAlreadyExistsException. + $client = $newClient([$alreadyExists, $notFound, null]); + try { + $addTasksIndividually->invoke(new PushQueue('default'), $tasks, $queueName, $client); + $this->fail('Expected TaskQueueException'); + } catch (TaskQueueException $e) { + $this->assertNotInstanceOf(TaskAlreadyExistsException::class, $e); + $this->assertEquals(3, $client->calls); + } + + $client = $newClient([null, null, null]); + $this->assertEquals(['a', 'b', 'c'], $addTasksIndividually->invoke( + new PushQueue('default'), $tasks, $queueName, $client)); + } + + public function testCloudTasksV2BatchCreateTasksPositionalSignatureAndTypeErrorPropagation() { + self::ensureCloudTasksV2StubsLoaded(); + self::setCloudTasksEnv(); + try { + $capturedParent = null; + PushQueue::setCloudTasksClientFactory(function ($className) use (&$capturedParent) { + return new class($capturedParent) { + private $parentRef; + public function __construct(&$ref) { $this->parentRef = &$ref; } + public function batchCreateTasks($parent, $requests) { + $this->parentRef = $parent; + return new class { + public function getMetadata() { + return new \Google\Cloud\Tasks\V2\BatchCreateTasksMetadata(); + } + public function getResult() { + $t1 = (new \Google\Cloud\Tasks\V2\Task()) + ->setName('projects/test-proj/locations/us-central1/queues/default/tasks/p1'); + $t2 = (new \Google\Cloud\Tasks\V2\Task()) + ->setName('projects/test-proj/locations/us-central1/queues/default/tasks/p2'); + return (new \Google\Cloud\Tasks\V2\BatchCreateTasksResponse()) + ->setTasks([$t1, $t2]); + } + }; + } + public function close() {} + }; + }); + + $queue = new PushQueue('default'); + $names = $queue->addTasks([new PushTask('/a'), new PushTask('/b')]); + $this->assertEquals(['p1', 'p2'], $names); + $this->assertEquals('projects/test-proj/locations/us-central1/queues/default', $capturedParent); + } finally { + self::clearCloudTasksEnv(); + } + } + + public function testCloudTasksV2BatchCreateTasksNotDoneThrowsWithoutPolling() { + self::ensureCloudTasksV2StubsLoaded(); + self::setCloudTasksEnv(); + try { + PushQueue::setCloudTasksClientFactory(function ($className) { + return new class { + public function batchCreateTasks($req) { + return new class { + public function isDone() { return false; } + public function pollUntilComplete() { + throw new \RuntimeException('pollUntilComplete should never be called'); + } + }; + } + public function close() {} + }; + }); + + $this->expectException(TaskQueueException::class); + $this->expectExceptionMessage('Cloud Tasks batch create operation returned done=false'); + $queue = new PushQueue('default'); + $queue->addTasks([new PushTask('/a'), new PushTask('/b')]); + } finally { + self::clearCloudTasksEnv(); + } + } + }