Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-05-25 12:26
📄 __init__.py # Copyright 2017, Google LLC 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. from __future__ import absolute_import from google.cloud.pubsub_v1.subscriber.client import Client __all__ = ("Client",)	721 B	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📁 _protocol	-	2025-05-25 12:26
📄 client.py # Copyright 2019, Google LLC 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. from __future__ import absolute_import import sys import os import typing from typing import cast, Any, Callable, Optional, Sequence, Union import warnings from google.auth.credentials import AnonymousCredentials # type: ignore from google.oauth2 import service_account # type: ignore from google.cloud.pubsub_v1 import types from google.cloud.pubsub_v1.subscriber import futures from google.cloud.pubsub_v1.subscriber._protocol import streaming_pull_manager from google.pubsub_v1.services.subscriber import client as subscriber_client from google.pubsub_v1 import gapic_version as package_version if typing.TYPE_CHECKING: # pragma: NO COVER from google.cloud.pubsub_v1 import subscriber from google.pubsub_v1.services.subscriber.transports.grpc import ( SubscriberGrpcTransport, ) __version__ = package_version.__version__ class Client(subscriber_client.SubscriberClient): """A subscriber client for Google Cloud Pub/Sub. This creates an object that is capable of subscribing to messages. Generally, you can instantiate this client with no arguments, and you get sensible defaults. Args: kwargs: Any additional arguments provided are sent as keyword keyword arguments to the underlying :class:`~google.cloud.pubsub_v1.gapic.subscriber_client.SubscriberClient`. Generally you should not need to set additional keyword arguments. Optionally, regional endpoints can be set via ``client_options`` that takes a single key-value pair that defines the endpoint. Example: .. code-block:: python from google.cloud import pubsub_v1 subscriber_client = pubsub_v1.SubscriberClient( # Optional client_options = { "api_endpoint": REGIONAL_ENDPOINT } ) """ def __init__( self, subscriber_options: Union[types.SubscriberOptions, Sequence] = (), kwargs: Any ): assert ( isinstance(subscriber_options, types.SubscriberOptions) or len(subscriber_options) == 0 ), "subscriber_options must be of type SubscriberOptions or an empty sequence." # Sanity check: Is our goal to use the emulator? # If so, create a grpc insecure channel with the emulator host # as the target. # TODO(https://github.com/googleapis/python-pubsub/issues/1349): Move the emulator # code below to test files. if os.environ.get("PUBSUB_EMULATOR_HOST"): kwargs["client_options"] = { "api_endpoint": os.environ.get("PUBSUB_EMULATOR_HOST") } # Configure credentials directly to transport, if provided. if "transport" not in kwargs: kwargs["credentials"] = AnonymousCredentials() # Instantiate the underlying GAPIC client. super().__init__(kwargs) self._target = self._transport._host self._closed = False self.subscriber_options = types.SubscriberOptions(subscriber_options) # Set / override Open Telemetry option. self._open_telemetry_enabled = ( self.subscriber_options.enable_open_telemetry_tracing ) # OpenTelemetry features used by the library are not supported in Python versions <= 3.7. # Refer https://github.com/open-telemetry/opentelemetry-python/issues/3993#issuecomment-2211976389 if ( self.subscriber_options.enable_open_telemetry_tracing and sys.version_info.major == 3 and sys.version_info.minor < 8 ): warnings.warn( message="Open Telemetry for Python version 3.7 or lower is not supported. Disabling Open Telemetry tracing.", category=RuntimeWarning, ) self._open_telemetry_enabled = False @property def open_telemetry_enabled(self) -> bool: """ Returns True if Open Telemetry is enabled. False otherwise. """ return self._open_telemetry_enabled # pragma: NO COVER @classmethod def from_service_account_file( # type: ignore[override] cls, filename: str, kwargs: Any ) -> "Client": """Creates an instance of this client using the provided credentials file. Args: filename: The path to the service account private key json file. kwargs: Additional arguments to pass to the constructor. Returns: A Subscriber :class:`~google.cloud.pubsub_v1.subscriber.client.Client` instance that is the constructed client. """ credentials = service_account.Credentials.from_service_account_file(filename) kwargs["credentials"] = credentials return cls(kwargs) from_service_account_json = from_service_account_file # type: ignore[assignment] @property def target(self) -> str: """Return the target (where the API is). Returns: The location of the API. """ return self._target @property def closed(self) -> bool: """Return whether the client has been closed and cannot be used anymore. .. versionadded:: 2.8.0 """ return self._closed @property def api(self): """The underlying gapic API client. .. versionchanged:: 2.10.0 Instead of a GAPIC ``SubscriberClient`` client instance, this property is a proxy object to it with the same interface. .. deprecated:: 2.10.0 Use the GAPIC methods and properties on the client instance directly instead of through the :attr:`api` attribute. """ msg = ( 'The "api" property only exists for backward compatibility, access its ' 'attributes directly thorugh the client instance (e.g. "client.foo" ' 'instead of "client.api.foo").' ) warnings.warn(msg, category=DeprecationWarning) return super() def subscribe( self, subscription: str, callback: Callable[["subscriber.message.Message"], Any], flow_control: Union[types.FlowControl, Sequence] = (), scheduler: Optional["subscriber.scheduler.ThreadScheduler"] = None, use_legacy_flow_control: bool = False, await_callbacks_on_shutdown: bool = False, ) -> futures.StreamingPullFuture: """Asynchronously start receiving messages on a given subscription. This method starts a background thread to begin pulling messages from a Pub/Sub subscription and scheduling them to be processed using the provided ``callback``. The ``callback`` will be called with an individual :class:`google.cloud.pubsub_v1.subscriber.message.Message`. It is the responsibility of the callback to either call ``ack()`` or ``nack()`` on the message when it finished processing. If an exception occurs in the callback during processing, the exception is logged and the message is ``nack()`` ed. The ``flow_control`` argument can be used to control the rate of at which messages are pulled. The settings are relatively conservative by default to prevent "message hoarding" - a situation where the client pulls a large number of messages but can not process them fast enough leading it to "starve" other clients of messages. Increasing these settings may lead to faster throughput for messages that do not take a long time to process. The ``use_legacy_flow_control`` argument disables enforcing flow control settings at the Cloud Pub/Sub server, and only the client side flow control will be enforced. This method starts the receiver in the background and returns a Future* representing its execution. Waiting on the future (calling ``result()``) will block forever or until a non-recoverable error is encountered (such as loss of network connectivity). Cancelling the future will signal the process to shutdown gracefully and exit. .. note:: This uses Pub/Sub's streaming pull feature. This feature properties that may be surprising. Please take a look at https://cloud.google.com/pubsub/docs/pull#streamingpull for more details on how streaming pull behaves compared to the synchronous pull method. Example: .. code-block:: python from google.cloud import pubsub_v1 subscriber_client = pubsub_v1.SubscriberClient() # existing subscription subscription = subscriber_client.subscription_path( 'my-project-id', 'my-subscription') def callback(message): print(message) message.ack() future = subscriber_client.subscribe( subscription, callback) try: future.result() except KeyboardInterrupt: future.cancel() # Trigger the shutdown. future.result() # Block until the shutdown is complete. Args: subscription: The name of the subscription. The subscription should have already been created (for example, by using :meth:`create_subscription`). callback: The callback function. This function receives the message as its only argument and will be called from a different thread/ process depending on the scheduling strategy. flow_control: The flow control settings. Use this to prevent situations where you are inundated with too many messages at once. scheduler: An optional scheduler to use when executing the callback. This controls how callbacks are executed concurrently. This object must not be shared across multiple ``SubscriberClient`` instances. use_legacy_flow_control (bool): If set to ``True``, flow control at the Cloud Pub/Sub server is disabled, though client-side flow control is still enabled. If set to ``False`` (default), both server-side and client-side flow control are enabled. await_callbacks_on_shutdown: If ``True``, after canceling the returned future, the latter's ``result()`` method will block until the background stream and its helper threads have been terminated, and all currently executing message callbacks are done processing. If ``False`` (default), the returned future's ``result()`` method will not block after canceling the future. The method will instead return immediately after the background stream and its helper threads have been terminated, but some of the message callback threads might still be running at that point. Returns: A future instance that can be used to manage the background stream. """ flow_control = types.FlowControl(*flow_control) manager = streaming_pull_manager.StreamingPullManager( self, subscription, flow_control=flow_control, scheduler=scheduler, use_legacy_flow_control=use_legacy_flow_control, await_callbacks_on_shutdown=await_callbacks_on_shutdown, ) future = futures.StreamingPullFuture(manager) manager.open(callback=callback, on_callback_error=future.set_exception) return future def close(self) -> None: """Close the underlying channel to release socket resources. After a channel has been closed, the client instance cannot be used anymore. This method is idempotent. """ transport = cast("SubscriberGrpcTransport", self._transport) transport.grpc_channel.close() self._closed = True def __enter__(self) -> "Client": if self._closed: raise RuntimeError("Closed subscriber cannot be used as context manager.") return self def __exit__(self, exc_type, exc_val, exc_tb): self.close()	12.71 KB	2025-05-25 12:26
📄 exceptions.py # Copyright 2017, Google LLC 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. from __future__ import absolute_import from enum import Enum from google.api_core.exceptions import GoogleAPICallError from typing import Optional class AcknowledgeStatus(Enum): SUCCESS = 1 PERMISSION_DENIED = 2 FAILED_PRECONDITION = 3 INVALID_ACK_ID = 4 OTHER = 5 class AcknowledgeError(GoogleAPICallError): """Error during ack/modack/nack operation on exactly-once-enabled subscription.""" def __init__(self, error_code: AcknowledgeStatus, info: Optional[str]): self.error_code = error_code self.info = info message = None if info: message = str(self.error_code) + " : " + str(self.info) else: message = str(self.error_code) super(AcknowledgeError, self).__init__(message) __all__ = ("AcknowledgeError",)	1.38 KB	2025-05-25 12:26
📄 futures.py # Copyright 2017, Google LLC 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. from __future__ import absolute_import import typing from typing import Any from typing import Union from google.cloud.pubsub_v1 import futures from google.cloud.pubsub_v1.subscriber.exceptions import AcknowledgeStatus if typing.TYPE_CHECKING: # pragma: NO COVER from google.cloud.pubsub_v1.subscriber._protocol.streaming_pull_manager import ( StreamingPullManager, ) class StreamingPullFuture(futures.Future): """Represents a process that asynchronously performs streaming pull and schedules messages to be processed. This future is resolved when the process is stopped (via :meth:`cancel`) or if it encounters an unrecoverable error. Calling `.result()` will cause the calling thread to block indefinitely. """ def __init__(self, manager: "StreamingPullManager"): super(StreamingPullFuture, self).__init__() self.__manager = manager self.__manager.add_close_callback(self._on_close_callback) self.__cancelled = False def _on_close_callback(self, manager: "StreamingPullManager", result: Any): if self.done(): # The future has already been resolved in a different thread, # nothing to do on the streaming pull manager shutdown. return if result is None: self.set_result(True) else: self.set_exception(result) def cancel(self) -> bool: """Stops pulling messages and shutdowns the background thread consuming messages. The method always returns ``True``, as the shutdown is always initiated. However, if the background stream is already being shut down or the shutdown has completed, this method is a no-op. .. versionchanged:: 2.4.1 The method does not block anymore, it just triggers the shutdown and returns immediately. To block until the background stream is terminated, call :meth:`result()` after cancelling the future. .. versionchanged:: 2.10.0 The method always returns ``True`` instead of ``None``. """ # NOTE: We circumvent the base future's self._state to track the cancellation # state, as this state has different meaning with streaming pull futures. self.__cancelled = True self.__manager.close() return True def cancelled(self) -> bool: """ Returns: ``True`` if the subscription has been cancelled. """ return self.__cancelled class Future(futures.Future): """This future object is for subscribe-side calls. Calling :meth:`result` will resolve the future by returning the message ID, unless an error occurs. """ def cancel(self) -> bool: """Actions in Pub/Sub generally may not be canceled. This method always returns ``False``. """ return False def cancelled(self) -> bool: """Actions in Pub/Sub generally may not be canceled. This method always returns ``False``. """ return False def result(self, timeout: Union[int, float, None] = None) -> AcknowledgeStatus: """Return a success code or raise an exception. This blocks until the operation completes successfully and returns the error code unless an exception is raised. Args: timeout: The number of seconds before this call times out and raises TimeoutError. Returns: AcknowledgeStatus.SUCCESS if the operation succeeded. Raises: concurrent.futures.TimeoutError: If the request times out. AcknowledgeError: If the operation did not succeed for another reason. """ return super().result(timeout=timeout)	4.29 KB	2025-05-25 12:26
📄 message.py # Copyright 2017, Google LLC 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. from __future__ import absolute_import import datetime as dt import json import math import time import typing from typing import Optional, Callable from google.cloud.pubsub_v1.subscriber._protocol import requests from google.cloud.pubsub_v1.subscriber import futures from google.cloud.pubsub_v1.subscriber.exceptions import AcknowledgeStatus from google.cloud.pubsub_v1.open_telemetry.subscribe_opentelemetry import ( SubscribeOpenTelemetry, ) if typing.TYPE_CHECKING: # pragma: NO COVER import datetime import queue from google.cloud.pubsub_v1 import types from google.protobuf.internal import containers _MESSAGE_REPR = """\ Message {{ data: {!r} ordering_key: {!r} attributes: {} }}""" _SUCCESS_FUTURE = futures.Future() _SUCCESS_FUTURE.set_result(AcknowledgeStatus.SUCCESS) def _indent(lines: str, prefix: str = " ") -> str: """Indent some text. Note that this is present as ``textwrap.indent``, but not in Python 2. Args: lines: The newline delimited string to be indented. prefix: The prefix to indent each line with. Defaults to two spaces. Returns: The newly indented content. """ indented = [] for line in lines.split("\n"): indented.append(prefix + line) return "\n".join(indented) class Message(object): """A representation of a single Pub/Sub message. The common way to interact with :class:`~.pubsub_v1.subscriber.message.Message` objects is to receive them in callbacks on subscriptions; most users should never have a need to instantiate them by hand. (The exception to this is if you are implementing a custom subclass to :class:`~.pubsub_v1.subscriber._consumer.Consumer`.) Attributes: message_id (str): The message ID. In general, you should not need to use this directly. data (bytes): The data in the message. Note that this will be a :class:`bytes`, not a text string. attributes (MutableMapping[str, str]): The attributes sent along with the message. See :attr:`attributes` for more information on this type. publish_time (google.protobuf.timestamp_pb2.Timestamp): The time that this message was originally published. opentelemetry_data (google.cloud.pubsub_v1.open_telemetry.subscribe_opentelemetry.SubscribeOpenTelemetry) Open Telemetry data associated with this message. None if Open Telemetry is not enabled. """ def __init__( self, message: "types.PubsubMessage._meta._pb", # type: ignore ack_id: str, delivery_attempt: int, request_queue: "queue.Queue", exactly_once_delivery_enabled_func: Callable[[], bool] = lambda: False, ): """Construct the Message. .. note:: This class should not be constructed directly; it is the responsibility of :class:`BasePolicy` subclasses to do so. Args: message (types.PubsubMessage._meta._pb): The message received from Pub/Sub. For performance reasons it should be the raw protobuf message normally wrapped by :class:`~pubsub_v1.types.PubsubMessage`. A raw message can be obtained from a :class:`~pubsub_v1.types.PubsubMessage` instance through the latter's ``._pb`` attribute. ack_id (str): The ack_id received from Pub/Sub. delivery_attempt (int): The delivery attempt counter received from Pub/Sub if a DeadLetterPolicy is set on the subscription, and zero otherwise. request_queue (queue.Queue): A queue provided by the policy that can accept requests; the policy is responsible for handling those requests. exactly_once_delivery_enabled_func (Callable[[], bool]): A Callable that returns whether exactly-once delivery is currently-enabled. Defaults to a lambda that always returns False. """ self._message = message self._ack_id = ack_id self._delivery_attempt = delivery_attempt if delivery_attempt > 0 else None self._request_queue = request_queue self._exactly_once_delivery_enabled_func = exactly_once_delivery_enabled_func self.message_id = message.message_id # The instantiation time is the time that this message # was received. Tracking this provides us a way to be smart about # the default lease deadline. self._received_timestamp = time.time() # Store the message attributes directly to speed up attribute access, i.e. # to avoid two lookups if self._message.<attribute> pattern was used in # properties. self._attributes = message.attributes self._data = message.data self._publish_time = dt.datetime.fromtimestamp( message.publish_time.seconds + message.publish_time.nanos / 1e9, tz=dt.timezone.utc, ) self._ordering_key = message.ordering_key self._size = message.ByteSize() # None if Open Telemetry is disabled. Else contains OpenTelemetry data. self._opentelemetry_data: Optional[SubscribeOpenTelemetry] = None def __repr__(self): # Get an abbreviated version of the data. abbv_data = self._message.data if len(abbv_data) > 50: abbv_data = abbv_data[:50] + b"..." pretty_attrs = json.dumps( dict(self.attributes), indent=2, separators=(",", ": "), sort_keys=True ) pretty_attrs = _indent(pretty_attrs) # We don't actually want the first line indented. pretty_attrs = pretty_attrs.lstrip() return _MESSAGE_REPR.format(abbv_data, str(self.ordering_key), pretty_attrs) @property def opentelemetry_data(self): return self._opentelemetry_data # pragma: NO COVER @opentelemetry_data.setter def opentelemetry_data(self, data): self._opentelemetry_data = data # pragma: NO COVER @property def attributes(self) -> "containers.ScalarMap": """Return the attributes of the underlying Pub/Sub Message. .. warning:: A ``ScalarMap`` behaves slightly differently than a ``dict``. For a Pub / Sub message this is a ``string->string`` map. When trying to access a value via ``map['key']``, if the key is not in the map, then the default value for the string type will be returned, which is an empty string. It may be more intuitive to just cast the map to a ``dict`` or to one use ``map.get``. Returns: containers.ScalarMap: The message's attributes. This is a ``dict``-like object provided by ``google.protobuf``. """ return self._attributes @property def data(self) -> bytes: """Return the data for the underlying Pub/Sub Message. Returns: bytes: The message data. This is always a bytestring; if you want a text string, call :meth:`bytes.decode`. """ return self._data @property def publish_time(self) -> "datetime.datetime": """Return the time that the message was originally published. Returns: datetime.datetime: The date and time that the message was published. """ return self._publish_time @property def ordering_key(self) -> str: """The ordering key used to publish the message.""" return self._ordering_key @property def size(self) -> int: """Return the size of the underlying message, in bytes.""" return self._size @property def ack_id(self) -> str: """the ID used to ack the message.""" return self._ack_id @property def delivery_attempt(self) -> Optional[int]: """The delivery attempt counter is 1 + (the sum of number of NACKs and number of ack_deadline exceeds) for this message. It is set to None if a DeadLetterPolicy is not set on the subscription. A NACK is any call to ModifyAckDeadline with a 0 deadline. An ack_deadline exceeds event is whenever a message is not acknowledged within ack_deadline. Note that ack_deadline is initially Subscription.ackDeadlineSeconds, but may get extended automatically by the client library. The first delivery of a given message will have this value as 1. The value is calculated at best effort and is approximate. Returns: Optional[int]: The delivery attempt counter or ``None``. """ return self._delivery_attempt def ack(self) -> None: """Acknowledge the given message. Acknowledging a message in Pub/Sub means that you are done with it, and it will not be delivered to this subscription again. You should avoid acknowledging messages until you have finished processing them, so that in the event of a failure, you receive the message again. .. warning:: Acks in Pub/Sub are best effort. You should always ensure that your processing code is idempotent, as you may receive any given message more than once. If you need strong guarantees about acks and re-deliveres, enable exactly-once delivery on your subscription and use the `ack_with_response` method instead. Exactly once delivery is a preview feature. For more details, see: https://cloud.google.com/pubsub/docs/exactly-once-delivery." """ if self.opentelemetry_data: self.opentelemetry_data.add_process_span_event("ack called") self.opentelemetry_data.end_process_span() time_to_ack = math.ceil(time.time() - self._received_timestamp) self._request_queue.put( requests.AckRequest( ack_id=self._ack_id, byte_size=self.size, time_to_ack=time_to_ack, ordering_key=self.ordering_key, future=None, opentelemetry_data=self.opentelemetry_data, ) ) def ack_with_response(self) -> "futures.Future": """Acknowledge the given message. Acknowledging a message in Pub/Sub means that you are done with it, and it will not be delivered to this subscription again. You should avoid acknowledging messages until you have finished processing them, so that in the event of a failure, you receive the message again. If exactly-once delivery is NOT enabled on the subscription, the future returns immediately with an AcknowledgeStatus.SUCCESS. Since acks in Cloud Pub/Sub are best effort when exactly-once delivery is disabled, the message may be re-delivered. Because re-deliveries are possible, you should ensure that your processing code is idempotent, as you may receive any given message more than once. If exactly-once delivery is enabled on the subscription, the future returned by this method tracks the state of acknowledgement operation. If the future completes successfully, the message is guaranteed NOT to be re-delivered. Otherwise, the future will contain an exception with more details about the failure and the message may be re-delivered. Exactly once delivery is a preview feature. For more details, see https://cloud.google.com/pubsub/docs/exactly-once-delivery." Returns: futures.Future: A :class:`~google.cloud.pubsub_v1.subscriber.futures.Future` instance that conforms to Python Standard library's :class:`~concurrent.futures.Future` interface (but not an instance of that class). Call `result()` to get the result of the operation; upon success, a pubsub_v1.subscriber.exceptions.AcknowledgeStatus.SUCCESS will be returned and upon an error, an pubsub_v1.subscriber.exceptions.AcknowledgeError exception will be thrown. """ if self.opentelemetry_data: self.opentelemetry_data.add_process_span_event("ack called") self.opentelemetry_data.end_process_span() req_future: Optional[futures.Future] if self._exactly_once_delivery_enabled_func(): future = futures.Future() req_future = future else: future = _SUCCESS_FUTURE req_future = None time_to_ack = math.ceil(time.time() - self._received_timestamp) self._request_queue.put( requests.AckRequest( ack_id=self._ack_id, byte_size=self.size, time_to_ack=time_to_ack, ordering_key=self.ordering_key, future=req_future, opentelemetry_data=self.opentelemetry_data, ) ) return future def drop(self) -> None: """Release the message from lease management. This informs the policy to no longer hold on to the lease for this message. Pub/Sub will re-deliver the message if it is not acknowledged before the existing lease expires. .. warning:: For most use cases, the only reason to drop a message from lease management is on `ack` or `nack`; this library automatically drop()s the message on `ack` or `nack`. You probably do not want to call this method directly. """ self._request_queue.put( requests.DropRequest( ack_id=self._ack_id, byte_size=self.size, ordering_key=self.ordering_key ) ) def modify_ack_deadline(self, seconds: int) -> None: """Resets the deadline for acknowledgement. New deadline will be the given value of seconds from now. The default implementation handles automatically modacking received messages for you; you should not need to manually deal with setting ack deadlines. The exception case is if you are implementing your own custom subclass of :class:`~.pubsub_v1.subcriber._consumer.Consumer`. Args: seconds (int): The number of seconds to set the lease deadline to. This should be between 0 and 600. Due to network latency, values below 10 are advised against. """ self._request_queue.put( requests.ModAckRequest( ack_id=self._ack_id, seconds=seconds, future=None, opentelemetry_data=self.opentelemetry_data, ) ) def modify_ack_deadline_with_response(self, seconds: int) -> "futures.Future": """Resets the deadline for acknowledgement and returns the response status via a future. New deadline will be the given value of seconds from now. The default implementation handles automatically modacking received messages for you; you should not need to manually deal with setting ack deadlines. The exception case is if you are implementing your own custom subclass of :class:`~.pubsub_v1.subcriber._consumer.Consumer`. If exactly-once delivery is NOT enabled on the subscription, the future returns immediately with an AcknowledgeStatus.SUCCESS. Since modify-ack-deadline operations in Cloud Pub/Sub are best effort when exactly-once delivery is disabled, the message may be re-delivered within the set deadline. If exactly-once delivery is enabled on the subscription, the future returned by this method tracks the state of the modify-ack-deadline operation. If the future completes successfully, the message is guaranteed NOT to be re-delivered within the new deadline. Otherwise, the future will contain an exception with more details about the failure and the message will be redelivered according to its currently-set ack deadline. Exactly once delivery is a preview feature. For more details, see https://cloud.google.com/pubsub/docs/exactly-once-delivery." Args: seconds (int): The number of seconds to set the lease deadline to. This should be between 0 and 600. Due to network latency, values below 10 are advised against. Returns: futures.Future: A :class:`~google.cloud.pubsub_v1.subscriber.futures.Future` instance that conforms to Python Standard library's :class:`~concurrent.futures.Future` interface (but not an instance of that class). Call `result()` to get the result of the operation; upon success, a pubsub_v1.subscriber.exceptions.AcknowledgeStatus.SUCCESS will be returned and upon an error, an pubsub_v1.subscriber.exceptions.AcknowledgeError exception will be thrown. """ req_future: Optional[futures.Future] if self._exactly_once_delivery_enabled_func(): future = futures.Future() req_future = future else: future = _SUCCESS_FUTURE req_future = None self._request_queue.put( requests.ModAckRequest( ack_id=self._ack_id, seconds=seconds, future=req_future, opentelemetry_data=self.opentelemetry_data, ) ) return future def nack(self) -> None: """Decline to acknowledge the given message. This will cause the message to be re-delivered to subscribers. Re-deliveries may take place immediately or after a delay, and may arrive at this subscriber or another. """ if self.opentelemetry_data: self.opentelemetry_data.add_process_span_event("nack called") self.opentelemetry_data.end_process_span() self._request_queue.put( requests.NackRequest( ack_id=self._ack_id, byte_size=self.size, ordering_key=self.ordering_key, future=None, opentelemetry_data=self.opentelemetry_data, ) ) def nack_with_response(self) -> "futures.Future": """Decline to acknowledge the given message, returning the response status via a future. This will cause the message to be re-delivered to subscribers. Re-deliveries may take place immediately or after a delay, and may arrive at this subscriber or another. If exactly-once delivery is NOT enabled on the subscription, the future returns immediately with an AcknowledgeStatus.SUCCESS. If exactly-once delivery is enabled on the subscription, the future returned by this method tracks the state of the nack operation. If the future completes successfully, the future's result will be an AcknowledgeStatus.SUCCESS. Otherwise, the future will contain an exception with more details about the failure. Exactly once delivery is a preview feature. For more details, see https://cloud.google.com/pubsub/docs/exactly-once-delivery." Returns: futures.Future: A :class:`~google.cloud.pubsub_v1.subscriber.futures.Future` instance that conforms to Python Standard library's :class:`~concurrent.futures.Future` interface (but not an instance of that class). Call `result()` to get the result of the operation; upon success, a pubsub_v1.subscriber.exceptions.AcknowledgeStatus.SUCCESS will be returned and upon an error, an pubsub_v1.subscriber.exceptions.AcknowledgeError exception will be thrown. """ if self.opentelemetry_data: self.opentelemetry_data.add_process_span_event("nack called") self.opentelemetry_data.end_process_span() req_future: Optional[futures.Future] if self._exactly_once_delivery_enabled_func(): future = futures.Future() req_future = future else: future = _SUCCESS_FUTURE req_future = None self._request_queue.put( requests.NackRequest( ack_id=self._ack_id, byte_size=self.size, ordering_key=self.ordering_key, future=req_future, opentelemetry_data=self.opentelemetry_data, ) ) return future	21.04 KB	2025-05-25 12:26
📄 scheduler.py # Copyright 2018, Google LLC # # 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 # # https://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. """Schedulers provide means to schedule callbacks asynchronously. These are used by the subscriber to call the user-provided callback to process each message. """ import abc import concurrent.futures import queue import typing from typing import Callable, List, Optional import warnings if typing.TYPE_CHECKING: # pragma: NO COVER from google.cloud import pubsub_v1 class Scheduler(metaclass=abc.ABCMeta): """Abstract base class for schedulers. Schedulers are used to schedule callbacks asynchronously. """ @property @abc.abstractmethod def queue(self) -> queue.Queue: # pragma: NO COVER """Queue: A concurrency-safe queue specific to the underlying concurrency implementation. This queue is used to send messages back to the scheduling actor. """ raise NotImplementedError @abc.abstractmethod def schedule(self, callback: Callable, args, kwargs) -> None: # pragma: NO COVER """Schedule the callback to be called asynchronously. Args: callback: The function to call. args: Positional arguments passed to the callback. kwargs: Key-word arguments passed to the callback. Returns: None """ raise NotImplementedError @abc.abstractmethod def shutdown( self, await_msg_callbacks: bool = False ) -> List["pubsub_v1.subscriber.message.Message"]: # pragma: NO COVER """Shuts down the scheduler and immediately end all pending callbacks. Args: await_msg_callbacks: If ``True``, the method will block until all currently executing callbacks are done processing. If ``False`` (default), the method will not wait for the currently running callbacks to complete. Returns: The messages submitted to the scheduler that were not yet dispatched to their callbacks. It is assumed that each message was submitted to the scheduler as the first positional argument to the provided callback. """ raise NotImplementedError def _make_default_thread_pool_executor() -> concurrent.futures.ThreadPoolExecutor: return concurrent.futures.ThreadPoolExecutor( max_workers=10, thread_name_prefix="ThreadPoolExecutor-ThreadScheduler" ) class ThreadScheduler(Scheduler): """A thread pool-based scheduler. It must not be shared across SubscriberClients. This scheduler is useful in typical I/O-bound message processing. Args: executor: An optional executor to use. If not specified, a default one will be created. """ def __init__( self, executor: Optional[concurrent.futures.ThreadPoolExecutor] = None ): self._queue: queue.Queue = queue.Queue() if executor is None: self._executor = _make_default_thread_pool_executor() else: self._executor = executor @property def queue(self): """Queue: A thread-safe queue used for communication between callbacks and the scheduling thread.""" return self._queue def schedule(self, callback: Callable, args, *kwargs) -> None: """Schedule the callback to be called asynchronously in a thread pool. Args: callback: The function to call. args: Positional arguments passed to the callback. kwargs: Key-word arguments passed to the callback. Returns: None """ try: self._executor.submit(callback, args, **kwargs) except RuntimeError: warnings.warn( "Scheduling a callback after executor shutdown.", category=RuntimeWarning, stacklevel=2, ) def shutdown( self, await_msg_callbacks: bool = False ) -> List["pubsub_v1.subscriber.message.Message"]: """Shut down the scheduler and immediately end all pending callbacks. Args: await_msg_callbacks: If ``True``, the method will block until all currently executing executor threads are done processing. If ``False`` (default), the method will not wait for the currently running threads to complete. Returns: The messages submitted to the scheduler that were not yet dispatched to their callbacks. It is assumed that each message was submitted to the scheduler as the first positional argument to the provided callback. """ dropped_messages = [] # Drop all pending item from the executor. Without this, the executor will also # try to process any pending work items before termination, which is undesirable. # # TODO: Replace the logic below by passing `cancel_futures=True` to shutdown() # once we only need to support Python 3.9+. try: while True: work_item = self._executor._work_queue.get(block=False) if work_item is None: # Exceutor in shutdown mode. continue dropped_messages.append(work_item.args[0]) # type: ignore[index] except queue.Empty: pass self._executor.shutdown(wait=await_msg_callbacks) return dropped_messages	5.86 KB	2025-05-25 12:26

Terminal