From cf064a63edc7a864d98fb07763e17ee34327b19b Mon Sep 17 00:00:00 2001 From: Riddhi Shivhare Date: Tue, 14 Apr 2026 10:46:50 +0530 Subject: [PATCH 1/2] CT integration using enviroment variable --- src/Api/TaskQueue/PushQueue.php | 104 ++++++++++++++++++++++++++++++++ src/Api/TaskQueue/PushTask.php | 30 +++++++++ 2 files changed, 134 insertions(+) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index fdfa520f..a3788b66 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -40,6 +40,9 @@ final class PushQueue { */ const MAX_TASKS_PER_ADD = 100; + const GAE_PUSHQUEUE_BACKEND = 'GAE_PUSHQUEUE_BACKEND'; + const CLOUD_TASK_BACKEND = 'CLOUD_TASK'; + private $name; private static $methods = [ @@ -147,6 +150,15 @@ public function addTasks($tasks) { '$tasks must contain at most ' . self::MAX_TASKS_PER_ADD . ' tasks. Actual size: ' . count($tasks)); } + + $backend = getenv(self::GAE_PUSHQUEUE_BACKEND); + if ($backend === self::CLOUD_TASK_BACKEND) { + if (count($tasks) > 1) { + throw new \RuntimeException('Batch operations are not supported for Cloud Tasks backend yet.'); + } + return [$this->createCloudTask($tasks[0])]; + } + $req = new TaskQueueBulkAddRequest(); $resp = new TaskQueueBulkAddResponse(); @@ -212,4 +224,96 @@ public function addTasks($tasks) { } return $names; } + + /** + * Create a task using Cloud Tasks API. + * + * @param PushTask $task The task to create. + * @return string The name of the created task. + * @throws TaskQueueException if there was a problem using the service. + */ + private function createCloudTask(PushTask $task) { + $projectId = \Google\AppEngine\Api\AppIdentity\AppIdentityService::getApplicationId(); + $location = getenv('GAE_LOCATION'); + if (!$location) { + throw new \RuntimeException('GAE_LOCATION environment variable is not set.'); + } + $queue = $this->name; + + $accessTokenData = \Google\AppEngine\Api\AppIdentity\AppIdentityService::getAccessToken(['https://www.googleapis.com/auth/cloud-platform']); + $token = $accessTokenData['access_token']; + + $url = sprintf('https://cloudtasks.googleapis.com/v2beta2/projects/%s/locations/%s/queues/%s/tasks', $projectId, $location, $queue); + + $client = new \GuzzleHttp\Client(); + + $body = [ + 'task' => [ + 'appEngineHttpRequest' => [ + 'httpMethod' => $task->getMethod(), + 'relativeUri' => $task->getUrl(), + ] + ] + ]; + + $retryConfig = $task->getRetryConfig(); + if (!empty($retryConfig)) { + $body['task']['retryConfig'] = []; + if (isset($retryConfig['max_attempts'])) { + $body['task']['retryConfig']['maxAttempts'] = $retryConfig['max_attempts']; + } + if (isset($retryConfig['min_backoff'])) { + $body['task']['retryConfig']['minBackoff'] = $retryConfig['min_backoff']['seconds'] . 's'; + } + if (isset($retryConfig['max_backoff'])) { + $body['task']['retryConfig']['maxBackoff'] = $retryConfig['max_backoff']['seconds'] . 's'; + } + } + + $headers = []; + foreach ($task->getHeaders() as $header) { + $pair = explode(':', $header, 2); + $headers[trim($pair[0])] = trim($pair[1]); + } + + if (!empty($headers)) { + $body['task']['appEngineHttpRequest']['headers'] = $headers; + } + + if ($task->getMethod() == 'POST' || $task->getMethod() == 'PUT') { + if ($task->getQueryData()) { + $body['task']['appEngineHttpRequest']['body'] = base64_encode(http_build_query($task->getQueryData())); + } + } + + if ($task->getName()) { + $body['task']['name'] = sprintf('projects/%s/locations/%s/queues/%s/tasks/%s', $projectId, $location, $queue, $task->getName()); + } + + if ($task->getDelaySeconds() > 0) { + $eta = time() + $task->getDelaySeconds(); + $body['task']['scheduleTime'] = gmdate('Y-m-d\TH:i:s\Z', $eta); + } + + try { + $response = $client->post($url, [ + 'headers' => [ + 'Authorization' => 'Bearer ' . $token, + 'Content-Type' => 'application/json', + ], + 'json' => $body, + ]); + + $respBody = json_decode($response->getBody(), true); + $fullName = $respBody['name']; + $parts = explode('/', $fullName); + return end($parts); + + } catch (\GuzzleHttp\Exception\RequestException $e) { + if ($e->getResponse() && $e->getResponse()->getStatusCode() == 409) { + throw new TaskAlreadyExistsException('Task with the same name exists already'); + } + throw new TaskQueueException('Cloud Tasks API error: ' . $e->getMessage()); + } + } } diff --git a/src/Api/TaskQueue/PushTask.php b/src/Api/TaskQueue/PushTask.php index e0409810..1c7233fe 100644 --- a/src/Api/TaskQueue/PushTask.php +++ b/src/Api/TaskQueue/PushTask.php @@ -64,6 +64,7 @@ final class PushTask { 'method' => 'POST', 'name' => '', 'header' => '', + 'retry_config' => [], ]; private $url; @@ -171,6 +172,26 @@ public function __construct($url_path, $query_data = [], $options = []) { ' (30 days). delay_seconds: ' . $delay); } + $retry_config = $this->options['retry_config']; + if (!is_array($retry_config)) { + throw new \InvalidArgumentException('retry_config must be an array.'); + } + if (!empty($retry_config)) { + if (isset($retry_config['max_attempts']) && !is_int($retry_config['max_attempts'])) { + throw new \InvalidArgumentException('max_attempts must be an integer.'); + } + if (isset($retry_config['min_backoff'])) { + if (!is_array($retry_config['min_backoff']) || !isset($retry_config['min_backoff']['seconds']) || !is_int($retry_config['min_backoff']['seconds'])) { + throw new \InvalidArgumentException('min_backoff must be an array with an integer "seconds" field.'); + } + } + if (isset($retry_config['max_backoff'])) { + if (!is_array($retry_config['max_backoff']) || !isset($retry_config['max_backoff']['seconds']) || !is_int($retry_config['max_backoff']['seconds'])) { + throw new \InvalidArgumentException('max_backoff must be an array with an integer "seconds" field.'); + } + } + } + $this->query_data = $query_data; $this->url = $url_path; if ($query_data) { @@ -275,6 +296,15 @@ public function getHeaders() { return $this->headers; } + /** + * Return the task's retry config. + * + * @return array The task's retry config. + */ + public function getRetryConfig() { + return $this->options['retry_config']; + } + /** * Adds the task to a queue. * From b09c85fb61b260b42c95e8d9a1773d077a13f535 Mon Sep 17 00:00:00 2001 From: Riddhi Shivhare Date: Tue, 14 Apr 2026 20:00:32 +0530 Subject: [PATCH 2/2] Minor Changes --- src/Api/TaskQueue/PushQueue.php | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/src/Api/TaskQueue/PushQueue.php b/src/Api/TaskQueue/PushQueue.php index a3788b66..d1e25521 100644 --- a/src/Api/TaskQueue/PushQueue.php +++ b/src/Api/TaskQueue/PushQueue.php @@ -45,6 +45,9 @@ final class PushQueue { private $name; + // The Guzzle Client used to make http requests. + private $guzzle_client = null; + private static $methods = [ 'POST' => RequestMethod::POST, 'GET' => RequestMethod::GET, @@ -57,14 +60,21 @@ final class PushQueue { * Construct a PushQueue * * @param string $name The name of the queue. + * @param \GuzzleHttp\Client $mock_client Mocked Guzzle Client used to make http requests. */ - public function __construct($name = 'default') { + public function __construct($name = 'default', $mock_client = null) { if (!is_string($name)) { throw new \InvalidArgumentException( '$name must be a string. Actual type: ' . gettype($name)); } # TODO: validate queue name length and regex. $this->name = $name; + + if (isset($mock_client)) { + $this->guzzle_client = $mock_client; + } else { + $this->guzzle_client = new \GuzzleHttp\Client(); + } } /** @@ -245,8 +255,6 @@ private function createCloudTask(PushTask $task) { $url = sprintf('https://cloudtasks.googleapis.com/v2beta2/projects/%s/locations/%s/queues/%s/tasks', $projectId, $location, $queue); - $client = new \GuzzleHttp\Client(); - $body = [ 'task' => [ 'appEngineHttpRequest' => [ @@ -296,7 +304,7 @@ private function createCloudTask(PushTask $task) { } try { - $response = $client->post($url, [ + $response = $this->guzzle_client->post($url, [ 'headers' => [ 'Authorization' => 'Bearer ' . $token, 'Content-Type' => 'application/json',