| Server IP : 34.87.99.72 / Your IP : 216.73.217.31 Web Server : Apache/2.4.46 (Unix) OpenSSL/1.1.1n System : Linux zlock-bitnami-lamp-prod-v2-vm 4.19.0-16-cloud-amd64 #1 SMP Debian 4.19.181-1 (2021-03-19) x86_64 User : bitnami ( 1000) PHP Version : 7.4.18 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : OFF Directory : /opt/bitnami/apache/htdocs/app/vendor/google/cloud-pubsub/src/ |
Upload File : |
<?php
/**
* Copyright 2017 Google Inc. All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
namespace Google\Cloud\PubSub;
use Google\Cloud\Core\ArrayTrait;
use Google\Cloud\Core\Batch\BatchRunner;
use Google\Cloud\Core\Batch\ClosureSerializerInterface;
use Google\Cloud\Core\Exception\NotFoundException;
use Google\Cloud\Core\Iam\Iam;
use Google\Cloud\Core\Iterator\ItemIterator;
use Google\Cloud\Core\Iterator\PageIterator;
use Google\Cloud\PubSub\Connection\ConnectionInterface;
use Google\Cloud\PubSub\Connection\IamTopic;
use Google\Cloud\PubSub\V1\Encoding;
use InvalidArgumentException;
/**
* A named resource to which messages are sent by publishers.
*
* Example:
* ```
* use Google\Cloud\PubSub\PubSubClient;
*
* $pubsub = new PubSubClient();
* $topic = $pubsub->topic('my-new-topic');
* ```
*
* ```
* // You can also pass a fully-qualified topic name:
* $topic = $pubsub->topic('projects/my-awesome-project/topics/my-new-topic');
* ```
*/
class Topic
{
use ArrayTrait;
use ResourceNameTrait;
const DEFAULT_COMPRESSION_BYTES_THRESHOLD = 240;
const DEFAULT_ENABLE_COMPRESSION = false;
/**
* @var ConnectionInterface A connection to the Google Cloud Platform API
* @internal
*/
protected $connection;
/**
* @var string The project ID
*/
private $projectId;
/**
* @var string The topic name
*/
private $name;
/**
* @var bool
*/
private $encode;
/**
* @var array
*/
private $info;
/**
* @var Iam
*/
private $iam;
/**
* @var array
*/
private $clientConfig;
/**
* @var bool
*/
private $enableCompression;
/**
* @var int
*/
private $compressionBytesThreshold;
/**
* Create a PubSub topic.
*
* @param ConnectionInterface $connection A connection to the Google Cloud
* Platform service. This object is created by PubSubClient,
* and should not be instantiated outside of this client.
* @param string $projectId The project Id
* @param string $name The topic name
* @param bool $encode Whether messages should be base64 encoded.
* @param array $info {
* A [Topic](https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.topics
*
* @type string name The name of the topic.
* @type array $labels Key value pairs used to organize your resources.
* @type string $kmsKeyName The resource name of the Cloud KMS
* CryptoKey to be used to protect access to messages published on this
* topic. The expected format is
* `projects/my-project/locations/kr-location/keyRings/my-kr/cryptoKeys/my-key`.
* @type bool $enableCompression Flag to enable compression of messages
* before publishing. Set the flag to `true` to enable compression.
* Defaults to `false`. Messsages are compressed if their total
* size >= `compressionBytesThreshold`, whose default value has
* been experimentally derived after performance evaluations.
* @type int $compressionBytesThreshold The threshold byte size
* above which messages are compressed. This only takes effect
* if `enableCompression` is set to `true`. Defaults to `240`.
* (This value is experiementally derived after performance evaluations.)
* }
*
* @param array $clientConfig [optional] Configuration options for the
* PubSub client used to handle processing of batch items through the
* daemon. For valid options please see
* {@see \Google\Cloud\PubSub\PubSubClient::__construct()}.
* **Defaults to** the options provided to the PubSub client
* associated with this instance.
*/
public function __construct(
ConnectionInterface $connection,
$projectId,
$name,
$encode,
array $info = [],
array $clientConfig = []
) {
$this->connection = $connection;
$this->projectId = $projectId;
$this->encode = (bool) $encode;
$this->info = $info;
$this->clientConfig = $clientConfig;
if (isset($info['enableCompression']) && $info['enableCompression'] === true) {
$this->enableCompression = true;
$this->compressionBytesThreshold = $info['compressionBytesThreshold'] ??
self::DEFAULT_COMPRESSION_BYTES_THRESHOLD;
} else {
$this->enableCompression = self::DEFAULT_ENABLE_COMPRESSION;
$this->compressionBytesThreshold = self::DEFAULT_COMPRESSION_BYTES_THRESHOLD;
}
// Accept either a simple name or a fully-qualified name.
if ($this->isFullyQualifiedName('topic', $name)) {
$this->name = $name;
} else {
$this->name = $this->formatName('topic', $name, $projectId);
}
}
/**
* Get the topic name
*
* Example:
* ```
* echo $topic->name();
* ```
*
* @return string
*/
public function name()
{
return $this->name;
}
/**
* Create a topic.
*
* Example:
* ```
* $topicInfo = $topic->create();
* ```
*
* @see https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.topics/create Create Topic
*
* @param array $options {
* Configuration Options
*
* @type array $labels Key value pairs used to organize your
* resources.
* @type string $kmsKeyName The resource name of the Cloud KMS
* CryptoKey to be used to protect access to messages published on
* this topic. The expected format is
* `projects/my-project/locations/kr-location/keyRings/my-kr/cryptoKeys/my-key`.
* @type array $messageStoragePolicy Policy constraining the set of
* Google Cloud Platform regions where messages published to the
* topic may be stored. If not present, then no constraints are in
* effect.
* @type string[] $messageStoragePolicy.allowedPersistenceRegions A list
* of IDs of GCP regions where messages that are published to the
* topic may be persisted in storage. Messages published by
* publishers running in non-allowed GCP regions (or running
* outside of GCP altogether) will be routed for storage in one of
* the allowed regions. An empty list means that no regions are
* allowed, and is not a valid configuration.
* @type string|Schema $schemaSettings.schema The name of a schema that
* messages published should be validated against, or an instance
* of {@see Schema}.
* @type string $schemaSettings.encoding The encoding of messages
* validated against schema. For allowed values, see constants
* defined on {@see Encoding}.
* }
*
* @return array Topic information
*/
public function create(array $options = [])
{
if (isset($options['schemaSettings']['schema']) && $options['schemaSettings']['schema'] instanceof Schema) {
$options['schemaSettings']['schema'] = $options['schemaSettings']['schema']->name();
}
$this->info = $this->connection->createTopic([
'name' => $this->name
] + $options);
return $this->info;
}
/**
* Update the topic.
*
* Note that the topic's name and kms key name are immutable properties and may not be modified.
*
* Example:
* ```
* $topic->update([
* 'messageStoragePolicy' => [
* 'allowedPersistenceRegions' => ['us-central1']
* ]
* ]);
* ```
*
* ```
* // Updating labels with an explicit update mask
* $topic->update([
* 'labels' => [
* 'foo' => 'bar'
* ]
* ], [
* 'updateMask' => [
* 'labels'
* ]
* ]);
* ```
*
* @see https://cloud.google.com/pubsub/docs/reference/rest/v1/UpdateTopicRequest Update Topic
*
* @param array $topic {
* The Topic data.
*
* @type array $labels Key value pairs used to organize your resources.
* @type array $messageStoragePolicy Policy constraining the set of
* Google Cloud Platform regions where messages published to the
* topic may be stored. If not present, then no constraints are in
* effect.
* @type string[] $messageStoragePolicy.allowedPersistenceRegions A list
* of IDs of GCP regions where messages that are published to the
* topic may be persisted in storage. Messages published by
* publishers running in non-allowed GCP regions (or running
* outside of GCP altogether) will be routed for storage in one of
* the allowed regions. An empty list means that no regions are
* allowed, and is not a valid configuration.
* @type string|Schema $schemaSettings.schema The name of a schema that
* messages published should be validated against, or an instance
* of {@see Schema}.
* @type string $schemaSettings.encoding The encoding of messages
* validated against schema. For allowed values, see constants
* defined on {@see V1\Encoding}.
* }
* @param array $options [optional] {
* Configuration options.
*
* @type array $updateMask A list of field paths to be modified. Nested
* key names should be dot-separated, e.g.
* `messageStoragePolicy.allowedPersistenceRegions`. Google Cloud
* PHP will attempt to infer this value on your behalf, however
* modification of map fields with arbitrary keys (such as labels
* or message storage policy) requires an explicit update mask.
* }
*
* @return array The topic info.
*/
public function update(array $topic, array $options = [])
{
$updateMaskPaths = $this->pluck('updateMask', $options, false) ?: [];
if (!$updateMaskPaths) {
$iterator = new \RecursiveIteratorIterator(new \RecursiveArrayIterator($topic));
foreach ($iterator as $leafValue) {
$excludes = ['name'];
$keys = [];
foreach (range(0, $iterator->getDepth()) as $depth) {
$key = $iterator->getSubIterator($depth)->key();
if (!is_string($key)) {
break;
}
$keys[] = $key;
}
$path = implode('.', $keys);
if (!in_array($path, $excludes)) {
$updateMaskPaths[] = $path;
}
}
}
if (isset($topic['schemaSettings']['schema'])) {
if ($topic['schemaSettings']['schema'] instanceof Schema) {
$topic['schemaSettings']['schema'] = $topic['schemaSettings']['schema']->name();
}
}
return $this->info = $this->connection->updateTopic([
'name' => $this->name,
'topic' => [
'name' => $this->name,
] + $topic,
'updateMask' => implode(',', $updateMaskPaths)
] + $options);
}
/**
* Delete a topic.
*
* Example:
* ```
* $topic->delete();
* ```
*
* @see https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.topics/delete Delete Topic
*
* @param array $options [optional] Configuration Options
* @return void
*/
public function delete(array $options = [])
{
$this->connection->deleteTopic($options + [
'topic' => $this->name
]);
}
/**
* Check if a topic exists.
*
* Service errors will NOT bubble up from this method. It will always return
* a boolean value. If you want to check for errors, use
* {@see Topic::info()}.
*
* Example:
* ```
* if ($topic->exists()) {
* echo 'Topic exists';
* }
* ```
*
* @param array $options [optional] Configuration Options
* @return bool
*/
public function exists(array $options = [])
{
try {
$this->info($options);
return true;
} catch (NotFoundException $e) {
return false;
}
}
/**
* Get topic information.
*
* Currently this resource returns only the topic name, if the topic exists.
* It is mostly useful therefore for checking if a topic exists.
*
* Since this method will throw an exception if the topic is not found, you
* may find that Topic::exists() is a better fit for a true/false check.
*
* This method will use the previously cached result, if available. To force
* a refresh from the API, use {@see Topic::reload()}.
*
* Example:
* ```
* $info = $topic->info();
* echo $info['name']; // projects/my-awesome-project/topics/my-new-topic
* ```
*
* @see https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.topics/get Get Topic
*
* @codingStandardsIgnoreStart
* @param array $options [optional] Configuration Options
* @return array [A representation of a Topic](https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.topics)
* @codingStandardsIgnoreEnd
*/
public function info(array $options = [])
{
if (!$this->info) {
$this->reload($options);
}
return $this->info;
}
/**
* Get topic information from the API.
*
* Currently this resource returns only the topic name, if the topic exists.
* It is mostly useful therefore for checking if a topic exists.
*
* Since this method will throw an exception if the topic is not found, you
* may find that Topic::exists() is a better fit for a true/false check.
*
* This method will retrieve a new result from the API. To use a previously
* cached result, if one exists, use {@see Topic::info()}.
*
* Example:
* ```
* $topic->reload();
* $info = $topic->info();
* echo $info['name']; // projects/my-awesome-project/topics/my-new-topic
* ```
*
* @see https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.topics/get Get Topic
*
* @codingStandardsIgnoreStart
* @param array $options [optional] Configuration Options
* @return array [A representation of a Topic](https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.topics)
* @codingStandardsIgnoreEnd
*/
public function reload(array $options = [])
{
return $this->info = $this->connection->getTopic($options + [
'topic' => $this->name
]);
}
/**
* Publish a new message to the topic.
*
* $message must provide at least one of `data` and `attributes` members.
*
* Example:
* ```
* $topic->publish([
* 'data' => 'New User Registered',
* 'attributes' => [
* 'id' => '1',
* 'userName' => 'John',
* 'location' => 'Detroit'
* ]
* ]);
* ```
*
* @see https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.topics/publish Publish Message
*
* @param Message|array $message An instance of
* {@see Message}, or an array in the correct
* [Message Format](https://cloud.google.com/pubsub/docs/reference/rest/v1/PubsubMessage).
* @param array $options [optional] Configuration Options
* @return array A list of message IDs
*/
public function publish($message, array $options = [])
{
return $this->publishBatch([$message], $options);
}
/**
* Publish multiple messages at once.
*
* Note that if ordering keys are provided, all members of given messages
* set must provide the same ordering key. Multiple ordering keys in a
* single publish call is not supported.
*
* Example:
* ```
* $topic->publishBatch([
* [
* 'data' => 'New User Registered',
* 'attributes' => [
* 'id' => '1',
* 'userName' => 'John',
* 'location' => 'Detroit'
* ]
* ], [
* 'data' => 'New User Registered',
* 'attributes' => [
* 'id' => '2',
* 'userName' => 'Steve',
* 'location' => 'Mountain View'
* ]
* ]
* ]);
* ```
*
* @see https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.topics/publish Publish Message
*
* @param Message[]|array[] $messages A list of messages. Each message must be in the correct
* [Message Format](https://cloud.google.com/pubsub/docs/reference/rest/v1/PubsubMessage),
* or be an instance of {@see Message}.
* @param array $options [optional] Configuration Options
* @return array A list of message IDs.
*/
public function publishBatch(array $messages, array $options = [])
{
foreach ($messages as &$message) {
$message = $this->formatMessage($message);
}
return $this->connection->publishMessage($options + [
'topic' => $this->name,
'messages' => $messages,
'compressionOptions' => [
'enableCompression' => $this->enableCompression,
'compressionBytesThreshold' => $this->compressionBytesThreshold
]
]);
}
/**
* Push a message into a batch queue, to be processed at a later point.
*
* Example:
* ```
* $topic->batchPublisher()
* ->publish([
* 'data' => 'New User Registered',
* 'attributes' => [
* 'id' => '2',
* 'userName' => 'Dave',
* 'location' => 'Detroit'
* ]
* ]);
* ```
*
* @param array $options [optional] {
* Configuration options.
*
* @type bool $debugOutput Whether or not to output debug information.
* Please note debug output currently only applies in CLI based
* applications. **Defaults to** `false`.
* @type resource $debugOutputResource A resource to output debug output
* to. **Defaults to** Resource for `php://stderr`.
* @type array $batchOptions A set of options for a BatchJob.
* {@see \Google\Cloud\Core\Batch\BatchJob::__construct()} for
* more details.
* **Defaults to** ['batchSize' => 1000,
* 'callPeriod' => 2.0,
* 'numWorkers' => 2].
* @type array $clientConfig Configuration options for the PubSub client
* used to handle processing of batch items. For valid options
* please see
* {@see \Google\Cloud\PubSub\PubSubClient::__construct()}.
* **Defaults to** the options provided to the client associated
* with the current `Topic` instance.
* @type BatchRunner $batchRunner A BatchRunner object. Mainly used for
* the tests to inject a mock. **Defaults to** a newly created
* BatchRunner.
* @type string $identifier An identifier for the batch job.
* **Defaults to** `pubsub-topic-{topic-name}`.
* Example: `pubsub-topic-mytopic`.
* @type ClosureSerializerInterface $closureSerializer An implementation
* responsible for serializing closures used in the
* `$clientConfig`. This is especially important when using the
* batch daemon. **Defaults to**
* {@see \Google\Cloud\Core\Batch\OpisClosureSerializer} if the
* `opis/closure` library is installed.
* @type bool $enableCompression Flag to enable compression of messages
* before publishing. Set the flag to `true` to enable compression.
* Defaults to `false`. Messsages are compressed if their total
* size >= `compressionBytesThreshold`, whose default value has
* been experimentally derived after performance evaluations.
* @type int $compressionBytesThreshold The threshold byte size
* above which messages are compressed. This only takes effect
* if `enableCompression` is set to `true`. Defaults to `240`.
* (This value is experiementally derived after performance evaluations.)
* }
* @return BatchPublisher
* @experimental The experimental flag means that while we believe this method
* or class is ready for use, it may change before release in backwards-
* incompatible ways. Please use with caution, and test thoroughly when
* upgrading.
*/
public function batchPublisher(array $options = [])
{
return new BatchPublisher(
$this->name,
$options + [
'clientConfig' => $this->clientConfig
]
);
}
/**
* Create a subscription to the topic.
*
* Example:
* ```
* $subscription = $topic->subscribe('my-new-subscription');
* ```
*
* @see https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.subscriptions/create Create Subscription
*
* @param string $name The subscription name
* @param array $options [optional] Please see {@see Subscription::create()}
* for configuration details.
* @return Subscription
*/
public function subscribe($name, array $options = [])
{
$subscription = $this->subscriptionFactory($name);
$subscription->create($options);
return $subscription;
}
/**
* This method will not run any API requests. You will receive a
* Subscription object that you can use to interact with the API.
*
* Example:
* ```
* $subscription = $topic->subscription('my-new-subscription');
* ```
*
* @param string $name The subscription name
* @return Subscription
*/
public function subscription($name)
{
return $this->subscriptionFactory($name);
}
/**
* Retrieve a list of active subscriptions to the current topic.
*
* Example:
* ```
* $subscriptions = $topic->subscriptions();
* foreach ($subscriptions as $subscription) {
* echo $subscription->name();
* }
* ```
*
* @codingStandardsIgnoreStart
* @see https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.topics.subscriptions/list List Topic Subscriptions
* @codingStandardsIgnoreEnd
*
* @param array $options [optional] {
* Configuration Options
*
* @type int $pageSize Maximum number of subscriptions to return.
* @type int $resultLimit Limit the number of results returned in total.
* **Defaults to** `0` (return all results).
* @type string $pageToken A previously-returned page token used to
* resume the loading of results from a specific point.
* }
* @return ItemIterator<Subscription>
*/
public function subscriptions(array $options = [])
{
$resultLimit = $this->pluck('resultLimit', $options, false);
return new ItemIterator(
new PageIterator(
function ($subscription) {
return $this->subscriptionFactory($subscription);
},
[$this->connection, 'listSubscriptionsByTopic'],
$options + ['topic' => $this->name],
[
'itemsKey' => 'subscriptions',
'resultLimit' => $resultLimit
]
)
);
}
/**
* Manage the IAM policy for the current Topic.
*
* Example:
* ```
* $iam = $topic->iam();
* ```
*
* @codingStandardsIgnoreStart
* @see https://cloud.google.com/pubsub/access_control PubSub Access Control Documentation
* @see https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.topics/getIamPolicy Get Topic IAM Policy
* @see https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.topics/setIamPolicy Set Topic IAM Policy
* @see https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.topics/testIamPermissions Test Topic Permissions
* @codingStandardsIgnoreEnd
*
* @return Iam
*/
public function iam()
{
if (!$this->iam) {
$iamConnection = new IamTopic($this->connection);
$this->iam = new Iam($iamConnection, $this->name);
}
return $this->iam;
}
/**
* Present a nicer debug result to people using php 5.6 or greater.
*
* @return array
* @codeCoverageIgnore
* @access private
*/
public function __debugInfo()
{
return [
'name' => $this->name,
'projectId' => $this->projectId,
'info' => $this->info,
'connection' => get_class($this->connection)
];
}
/**
* Ensure that the message is in a correct format,
* base64_encode the data, and error if the input is too wrong to proceed.
*
* @param Message|array $message
* @return array The message data
* @throws \InvalidArgumentException
*/
private function formatMessage($message)
{
$message = $message instanceof Message
? $message->toArray()
: $message;
if (isset($message['data']) && $this->encode) {
$message['data'] = base64_encode($message['data']);
}
if (!array_key_exists('data', $message) && !array_key_exists('attributes', $message)) {
throw new InvalidArgumentException(
'At least one of $data or $attributes must be specified on each
message, but neither was given.'
);
}
return $message;
}
/**
* Create a new subscription instance with the given name and optional
* subscription data.
*
* @codingStandardsIgnoreStart
* @param string $name
* @param array $info [optional] A representation of a
* [Subscription](https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.subscriptions#Subscription)
* @return Subscription
* @codingStandardsIgnoreEnd
*/
private function subscriptionFactory($name, array $info = [])
{
return new Subscription(
$this->connection,
$this->projectId,
$name,
$this->name,
$this->encode,
$info
);
}
}