Skip to content

CT integration using enviroment variable #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
riddhi-shivhare wants to merge 2 commits into
base: master
Choose a base branch
from
Open

CT integration using enviroment variable #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
cf064a6
CT integration using enviroment variable
riddhi-shivhare Apr 14, 2026
b09c85f
Minor Changes
riddhi-shivhare Apr 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
114 changes: 113 additions & 1 deletion src/Api/TaskQueue/PushQueue.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,8 +40,14 @@ final class PushQueue {
*/
const MAX_TASKS_PER_ADD = 100;

const GAE_PUSHQUEUE_BACKEND = 'GAE_PUSHQUEUE_BACKEND';
const CLOUD_TASK_BACKEND = 'CLOUD_TASK';

private $name;

// The Guzzle Client used to make http requests.
private $guzzle_client = null;

private static $methods = [
'POST' => RequestMethod::POST,
'GET' => RequestMethod::GET,
Expand All @@ -54,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();
}
}

/**
Expand Down Expand Up @@ -147,6 +160,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();

Expand Down Expand Up @@ -212,4 +234,94 @@ 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);

$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 = $this->guzzle_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());
}
}
}
30 changes: 30 additions & 0 deletions src/Api/TaskQueue/PushTask.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,7 @@ final class PushTask {
'method' => 'POST',
'name' => '',
'header' => '',
'retry_config' => [],

@gajjarc gajjarc Apr 15, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why are we adding this? It seems php doesn't support retry task at all? Can you double check?

];

private $url;
Expand Down Expand Up @@ -171,6 +172,26 @@ public function __construct($url_path, $query_data = [], $options = []) {
' (30 days). delay_seconds: ' . $delay);
}

$retry_config = $this->options['retry_config'];

@gajjarc gajjarc Apr 15, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It seems this is for all backends, TaskQueue and CloudTask. If that is the case why do we need to change the existing TaskQueue behaviour?

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) {
Expand Down Expand Up @@ -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.
*
Expand Down