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.publisher.client import Client __all__ = ("Client",)	720 B	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📁 _batch	-	2025-05-25 12:26
📁 _sequencer	-	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 copy import logging import os import threading import time import typing from typing import Any, Dict, Optional, Sequence, Tuple, Type, Union import warnings import sys from google.api_core import gapic_v1 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.publisher import exceptions from google.cloud.pubsub_v1.publisher import futures from google.cloud.pubsub_v1.publisher._batch import thread from google.cloud.pubsub_v1.publisher._sequencer import ordered_sequencer from google.cloud.pubsub_v1.publisher._sequencer import unordered_sequencer from google.cloud.pubsub_v1.publisher.flow_controller import FlowController from google.pubsub_v1 import gapic_version as package_version from google.pubsub_v1 import types as gapic_types from google.pubsub_v1.services.publisher import client as publisher_client from google.cloud.pubsub_v1.open_telemetry.publish_message_wrapper import ( PublishMessageWrapper, ) __version__ = package_version.__version__ if typing.TYPE_CHECKING: # pragma: NO COVER from google.cloud import pubsub_v1 from google.cloud.pubsub_v1.publisher import _batch from google.pubsub_v1.services.publisher.client import OptionalRetry from google.pubsub_v1.types import pubsub as pubsub_types _LOGGER = logging.getLogger(__name__) _raw_proto_pubbsub_message = gapic_types.PubsubMessage.pb() SequencerType = Union[ ordered_sequencer.OrderedSequencer, unordered_sequencer.UnorderedSequencer ] class Client(publisher_client.PublisherClient): """A publisher client for Google Cloud Pub/Sub. This creates an object that is capable of publishing messages. Generally, you can instantiate this client with no arguments, and you get sensible defaults. Args: batch_settings: The settings for batch publishing. publisher_options: The options for the publisher client. Note that enabling message ordering will override the publish retry timeout to be infinite. kwargs: Any additional arguments provided are sent as keyword arguments to the underlying :class:`~google.cloud.pubsub_v1.gapic.publisher_client.PublisherClient`. Generally you should not need to set additional keyword arguments. 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 publisher_client = pubsub_v1.PublisherClient( # Optional batch_settings = pubsub_v1.types.BatchSettings( max_bytes=1024, # One kilobyte max_latency=1, # One second ), # Optional publisher_options = pubsub_v1.types.PublisherOptions( enable_message_ordering=False, flow_control=pubsub_v1.types.PublishFlowControl( message_limit=2000, limit_exceeded_behavior=pubsub_v1.types.LimitExceededBehavior.BLOCK, ), ), # Optional client_options = { "api_endpoint": REGIONAL_ENDPOINT } ) """ def __init__( self, batch_settings: Union[types.BatchSettings, Sequence] = (), publisher_options: Union[types.PublisherOptions, Sequence] = (), *kwargs: Any, ): assert ( type(batch_settings) is types.BatchSettings or len(batch_settings) == 0 ), "batch_settings must be of type BatchSettings or an empty sequence." assert ( type(publisher_options) is types.PublisherOptions or len(publisher_options) == 0 ), "publisher_options must be of type PublisherOptions 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() # For a transient failure, retry publishing the message infinitely. self.publisher_options = types.PublisherOptions(publisher_options) self._enable_message_ordering = self.publisher_options[0] # Add the metrics headers, and instantiate the underlying GAPIC # client. super().__init__(*kwargs) self._target = self._transport._host self._batch_class = thread.Batch self.batch_settings = types.BatchSettings(batch_settings) # The batches on the publisher client are responsible for holding # messages. One batch exists for each topic. self._batch_lock = self._batch_class.make_lock() # (topic, ordering_key) => sequencers object self._sequencers: Dict[Tuple[str, str], SequencerType] = {} self._is_stopped = False # Thread created to commit all sequencers after a timeout. self._commit_thread: Optional[threading.Thread] = None # The object controlling the message publishing flow self._flow_controller = FlowController(self.publisher_options.flow_control) self._open_telemetry_enabled = ( self.publisher_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.publisher_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 @classmethod def from_service_account_file( # type: ignore[override] cls, filename: str, batch_settings: Union[types.BatchSettings, Sequence] = (), 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. batch_settings: The settings for batch publishing. kwargs: Additional arguments to pass to the constructor. Returns: A Publisher instance that is the constructed client. """ credentials = service_account.Credentials.from_service_account_file(filename) kwargs["credentials"] = credentials return cls(batch_settings, 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 api(self): """The underlying gapic API client. .. versionchanged:: 2.10.0 Instead of a GAPIC ``PublisherClient`` 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() @property def open_telemetry_enabled(self) -> bool: return self._open_telemetry_enabled def _get_or_create_sequencer(self, topic: str, ordering_key: str) -> SequencerType: """Get an existing sequencer or create a new one given the (topic, ordering_key) pair. """ sequencer_key = (topic, ordering_key) sequencer = self._sequencers.get(sequencer_key) if sequencer is None: if ordering_key == "": sequencer = unordered_sequencer.UnorderedSequencer(self, topic) else: sequencer = ordered_sequencer.OrderedSequencer( self, topic, ordering_key ) self._sequencers[sequencer_key] = sequencer return sequencer def resume_publish(self, topic: str, ordering_key: str) -> None: """Resume publish on an ordering key that has had unrecoverable errors. Args: topic: The topic to publish messages to. ordering_key: A string that identifies related messages for which publish order should be respected. Raises: RuntimeError: If called after publisher has been stopped by a `stop()` method call. ValueError: If the topic/ordering key combination has not been seen before by this client. """ with self._batch_lock: if self._is_stopped: raise RuntimeError("Cannot resume publish on a stopped publisher.") if not self._enable_message_ordering: raise ValueError( "Cannot resume publish on a topic/ordering key if ordering " "is not enabled." ) sequencer_key = (topic, ordering_key) sequencer = self._sequencers.get(sequencer_key) if sequencer is None: _LOGGER.debug( "Error: The topic/ordering key combination has not " "been seen before." ) else: sequencer.unpause() def _gapic_publish(self, args, kwargs) -> "pubsub_types.PublishResponse": """Call the GAPIC public API directly.""" return super().publish(args, kwargs) def publish( # type: ignore[override] self, topic: str, data: bytes, ordering_key: str = "", retry: "OptionalRetry" = gapic_v1.method.DEFAULT, timeout: "types.OptionalTimeout" = gapic_v1.method.DEFAULT, attrs: Union[bytes, str], ) -> "pubsub_v1.publisher.futures.Future": """Publish a single message. .. note:: Messages in Pub/Sub are blobs of bytes. They are binary data, not text. You must send data as a bytestring (``bytes`` in Python 3; ``str`` in Python 2), and this library will raise an exception if you send a text string. The reason that this is so important (and why we do not try to coerce for you) is because Pub/Sub is also platform independent and there is no way to know how to decode messages properly on the other side; therefore, encoding and decoding is a required exercise for the developer. Add the given message to this object; this will cause it to be published once the batch either has enough messages or a sufficient period of time has elapsed. This method may block if LimitExceededBehavior.BLOCK is used in the flow control settings. Example: >>> from google.cloud import pubsub_v1 >>> client = pubsub_v1.PublisherClient() >>> topic = client.topic_path('[PROJECT]', '[TOPIC]') >>> data = b'The rain in Wales falls mainly on the snails.' >>> response = client.publish(topic, data, username='guido') Args: topic: The topic to publish messages to. data: A bytestring representing the message body. This must be a bytestring. ordering_key: A string that identifies related messages for which publish order should be respected. Message ordering must be enabled for this client to use this feature. retry: Designation of what errors, if any, should be retried. If `ordering_key` is specified, the total retry deadline will be changed to "infinity". If given, it overides any retry passed into the client through the ``publisher_options`` argument. timeout: The timeout for the RPC request. Can be used to override any timeout passed in through ``publisher_options`` when instantiating the client. attrs: A dictionary of attributes to be sent as metadata. (These may be text strings or byte strings.) Returns: A :class:`~google.cloud.pubsub_v1.publisher.futures.Future` instance that conforms to Python Standard library's :class:`~concurrent.futures.Future` interface (but not an instance of that class). Raises: RuntimeError: If called after publisher has been stopped by a `stop()` method call. pubsub_v1.publisher.exceptions.MessageTooLargeError: If publishing the ``message`` would exceed the max size limit on the backend. """ # Sanity check: Is the data being sent as a bytestring? # If it is literally anything else, complain loudly about it. if not isinstance(data, bytes): raise TypeError( "Data being published to Pub/Sub must be sent as a bytestring." ) if not self._enable_message_ordering and ordering_key != "": raise ValueError( "Cannot publish a message with an ordering key when message " "ordering is not enabled." ) # Coerce all attributes to text strings. for k, v in copy.copy(attrs).items(): if isinstance(v, str): continue if isinstance(v, bytes): attrs[k] = v.decode("utf-8") continue raise TypeError( "All attributes being published to Pub/Sub must " "be sent as text strings." ) # Create the Pub/Sub message object. For performance reasons, the message # should be constructed by directly using the raw protobuf class, and only # then wrapping it into the higher-level PubsubMessage class. vanilla_pb = _raw_proto_pubbsub_message( data=data, ordering_key=ordering_key, attributes=attrs ) message = gapic_types.PubsubMessage.wrap(vanilla_pb) wrapper: PublishMessageWrapper = PublishMessageWrapper(message) if self._open_telemetry_enabled: wrapper.start_create_span(topic=topic, ordering_key=ordering_key) # Messages should go through flow control to prevent excessive # queuing on the client side (depending on the settings). try: if self._open_telemetry_enabled: if wrapper: wrapper.start_publisher_flow_control_span() else: # pragma: NO COVER warnings.warn( message="PubSubMessageWrapper is None. Not starting publisher flow control span.", category=RuntimeWarning, ) self._flow_controller.add(message) if self._open_telemetry_enabled: if wrapper: wrapper.end_publisher_flow_control_span() else: # pragma: NO COVER warnings.warn( message="PubSubMessageWrapper is None. Not ending publisher flow control span.", category=RuntimeWarning, ) except exceptions.FlowControlLimitError as exc: if self._open_telemetry_enabled: if wrapper: wrapper.end_publisher_flow_control_span(exc) wrapper.end_create_span(exc) else: # pragma: NO COVER warnings.warn( message="PubSubMessageWrapper is None. Not ending publisher create and flow control spans on FlowControlLimitError.", category=RuntimeWarning, ) future = futures.Future() future.set_exception(exc) return future def on_publish_done(future): self._flow_controller.release(message) if retry is gapic_v1.method.DEFAULT: # if custom retry not passed in retry = self.publisher_options.retry if timeout is gapic_v1.method.DEFAULT: # if custom timeout not passed in timeout = self.publisher_options.timeout if self._open_telemetry_enabled: if wrapper: wrapper.start_publisher_batching_span() else: # pragma: NO COVER warnings.warn( message="PublishMessageWrapper is None. Hence, not starting publisher batching span", category=RuntimeWarning, ) with self._batch_lock: try: if self._is_stopped: raise RuntimeError("Cannot publish on a stopped publisher.") # Set retry timeout to "infinite" when message ordering is enabled. # Note that this then also impacts messages added with an empty # ordering key. if self._enable_message_ordering: if retry is gapic_v1.method.DEFAULT: # use the default retry for the publish GRPC method as a base transport = self._transport base_retry = transport._wrapped_methods[ transport.publish ]._retry retry = base_retry.with_deadline(2.032) # timeout needs to be overridden and set to infinite in # addition to the retry deadline since both determine # the duration for which retries are attempted. timeout = 2.032 elif retry is not None: retry = retry.with_deadline(2.032) timeout = 2.032 # Delegate the publishing to the sequencer. sequencer = self._get_or_create_sequencer(topic, ordering_key) future = sequencer.publish( wrapper=wrapper, retry=retry, timeout=timeout ) future.add_done_callback(on_publish_done) except BaseException as be: # Exceptions can be thrown when attempting to add messages to # the batch. If they're thrown, record them in publisher # batching and create span, end the spans and bubble the # exception up. if self._open_telemetry_enabled: if wrapper: wrapper.end_publisher_batching_span(be) wrapper.end_create_span(be) else: # pragma: NO COVER warnings.warn( message="PublishMessageWrapper is None. Hence, not recording exception and ending publisher batching span and create span", category=RuntimeWarning, ) raise be if self._open_telemetry_enabled: if wrapper: wrapper.end_publisher_batching_span() else: # pragma: NO COVER warnings.warn( message="PublishMessageWrapper is None. Hence, not ending publisher batching span", category=RuntimeWarning, ) # Create a timer thread if necessary to enforce the batching # timeout. self._ensure_commit_timer_runs_no_lock() return future def ensure_cleanup_and_commit_timer_runs(self) -> None: """Ensure a cleanup/commit timer thread is running. If a cleanup/commit timer thread is already running, this does nothing. """ with self._batch_lock: self._ensure_commit_timer_runs_no_lock() def _ensure_commit_timer_runs_no_lock(self) -> None: """Ensure a commit timer thread is running, without taking _batch_lock. _batch_lock must be held before calling this method. """ if not self._commit_thread and self.batch_settings.max_latency < float("inf"): self._start_commit_thread() def _start_commit_thread(self) -> None: """Start a new thread to actually wait and commit the sequencers.""" # NOTE: If the thread is not a daemon, a memory leak exists due to a CPython issue. # https://github.com/googleapis/python-pubsub/issues/395#issuecomment-829910303 # https://github.com/googleapis/python-pubsub/issues/395#issuecomment-830092418 self._commit_thread = threading.Thread( name="Thread-PubSubBatchCommitter", target=self._wait_and_commit_sequencers, daemon=True, ) self._commit_thread.start() def _wait_and_commit_sequencers(self) -> None: """Wait up to the batching timeout, and commit all sequencers.""" # Sleep for however long we should be waiting. time.sleep(self.batch_settings.max_latency) _LOGGER.debug("Commit thread is waking up") with self._batch_lock: if self._is_stopped: return self._commit_sequencers() self._commit_thread = None def _commit_sequencers(self) -> None: """Clean up finished sequencers and commit the rest.""" finished_sequencer_keys = [ key for key, sequencer in self._sequencers.items() if sequencer.is_finished() ] for sequencer_key in finished_sequencer_keys: del self._sequencers[sequencer_key] for sequencer in self._sequencers.values(): sequencer.commit() def stop(self) -> None: """Immediately publish all outstanding messages. Asynchronously sends all outstanding messages and prevents future calls to `publish()`. Method should be invoked prior to deleting this `Client()` object in order to ensure that no pending messages are lost. .. note:: This method is non-blocking. Use `Future()` objects returned by `publish()` to make sure all publish requests completed, either in success or error. Raises: RuntimeError: If called after publisher has been stopped by a `stop()` method call. """ with self._batch_lock: if self._is_stopped: raise RuntimeError("Cannot stop a publisher already stopped.") self._is_stopped = True for sequencer in self._sequencers.values(): sequencer.stop() # Used only for testing. def _set_batch( self, topic: str, batch: "_batch.thread.Batch", ordering_key: str = "" ) -> None: sequencer = self._get_or_create_sequencer(topic, ordering_key) sequencer._set_batch(batch) # Used only for testing. def _set_batch_class(self, batch_class: Type) -> None: self._batch_class = batch_class # Used only for testing. def _set_sequencer( self, topic: str, sequencer: SequencerType, ordering_key: str = "" ) -> None: sequencer_key = (topic, ordering_key) self._sequencers[sequencer_key] = sequencer	24.6 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 google.api_core.exceptions import GoogleAPICallError from google.cloud.pubsub_v1.exceptions import TimeoutError class PublishError(GoogleAPICallError): pass class MessageTooLargeError(ValueError): """Attempt to publish a message that would exceed the server max size limit.""" class PublishToPausedOrderingKeyException(Exception): """Publish attempted to paused ordering key. To resume publishing, call the resumePublish method on the publisher Client object with this ordering key. Ordering keys are paused if an unrecoverable error occurred during publish of a batch for that key. """ def __init__(self, ordering_key: str): self.ordering_key = ordering_key super(PublishToPausedOrderingKeyException, self).__init__() class FlowControlLimitError(Exception): """An action resulted in exceeding the flow control limits.""" __all__ = ( "FlowControlLimitError", "MessageTooLargeError", "PublishError", "TimeoutError", "PublishToPausedOrderingKeyException", )	1.64 KB	2025-05-25 12:26
📄 flow_controller.py # Copyright 2020, 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 collections import OrderedDict import logging import threading from typing import Dict, Optional, Type import warnings from google.cloud.pubsub_v1 import types from google.cloud.pubsub_v1.publisher import exceptions _LOGGER = logging.getLogger(__name__) MessageType = Type[types.PubsubMessage] # type: ignore class _QuantityReservation: """A (partial) reservation of quantifiable resources.""" def __init__(self, bytes_reserved: int, bytes_needed: int, has_slot: bool): self.bytes_reserved = bytes_reserved self.bytes_needed = bytes_needed self.has_slot = has_slot def __repr__(self): return ( f"{type(self).__name__}(" f"bytes_reserved={self.bytes_reserved}, " f"bytes_needed={self.bytes_needed}, " f"has_slot={self.has_slot})" ) class FlowController(object): """A class used to control the flow of messages passing through it. Args: settings: Desired flow control configuration. """ def __init__(self, settings: types.PublishFlowControl): self._settings = settings # Load statistics. They represent the number of messages added, but not # yet released (and their total size). self._message_count = 0 self._total_bytes = 0 # A FIFO queue of threads blocked on adding a message that also tracks their # reservations of available flow control bytes and message slots. # Only relevant if the configured limit exceeded behavior is BLOCK. self._waiting: Dict[threading.Thread, _QuantityReservation] = OrderedDict() self._reserved_bytes = 0 self._reserved_slots = 0 # The lock is used to protect all internal state (message and byte count, # waiting threads to add, etc.). self._operational_lock = threading.Lock() # The condition for blocking the flow if capacity is exceeded. self._has_capacity = threading.Condition(lock=self._operational_lock) def add(self, message: MessageType) -> None: """Add a message to flow control. Adding a message updates the internal load statistics, and an action is taken if these limits are exceeded (depending on the flow control settings). Args: message: The message entering the flow control. Raises: :exception:`~pubsub_v1.publisher.exceptions.FlowControlLimitError`: Raised when the desired action is :attr:`~google.cloud.pubsub_v1.types.LimitExceededBehavior.ERROR` and the message would exceed flow control limits, or when the desired action is :attr:`~google.cloud.pubsub_v1.types.LimitExceededBehavior.BLOCK` and the message would block forever against the flow control limits. """ if self._settings.limit_exceeded_behavior == types.LimitExceededBehavior.IGNORE: return with self._operational_lock: if not self._would_overflow(message): self._message_count += 1 self._total_bytes += message._pb.ByteSize() return # Adding a message would overflow, react. if ( self._settings.limit_exceeded_behavior == types.LimitExceededBehavior.ERROR ): # Raising an error means rejecting a message, thus we do not # add anything to the existing load, but we do report the would-be # load if we accepted the message. load_info = self._load_info( message_count=self._message_count + 1, total_bytes=self._total_bytes + message._pb.ByteSize(), ) error_msg = "Flow control limits would be exceeded - {}.".format( load_info ) raise exceptions.FlowControlLimitError(error_msg) assert ( self._settings.limit_exceeded_behavior == types.LimitExceededBehavior.BLOCK ) # Sanity check - if a message exceeds total flow control limits all # by itself, it would block forever, thus raise error. if ( message._pb.ByteSize() > self._settings.byte_limit or self._settings.message_limit < 1 ): load_info = self._load_info( message_count=1, total_bytes=message._pb.ByteSize() ) error_msg = ( "Total flow control limits too low for the message, " "would block forever - {}.".format(load_info) ) raise exceptions.FlowControlLimitError(error_msg) current_thread = threading.current_thread() while self._would_overflow(message): if current_thread not in self._waiting: reservation = _QuantityReservation( bytes_reserved=0, bytes_needed=message._pb.ByteSize(), has_slot=False, ) self._waiting[current_thread] = reservation # Will be placed last. _LOGGER.debug( "Blocking until there is enough free capacity in the flow - " "{}.".format(self._load_info()) ) self._has_capacity.wait() _LOGGER.debug( "Woke up from waiting on free capacity in the flow - " "{}.".format(self._load_info()) ) # Message accepted, increase the load and remove thread stats. self._message_count += 1 self._total_bytes += message._pb.ByteSize() self._reserved_bytes -= self._waiting[current_thread].bytes_reserved self._reserved_slots -= 1 del self._waiting[current_thread] def release(self, message: MessageType) -> None: """Release a mesage from flow control. Args: message: The message entering the flow control. """ if self._settings.limit_exceeded_behavior == types.LimitExceededBehavior.IGNORE: return with self._operational_lock: # Releasing a message decreases the load. self._message_count -= 1 self._total_bytes -= message._pb.ByteSize() if self._message_count < 0 or self._total_bytes < 0: warnings.warn( "Releasing a message that was never added or already released.", category=RuntimeWarning, stacklevel=2, ) self._message_count = max(0, self._message_count) self._total_bytes = max(0, self._total_bytes) self._distribute_available_capacity() # If at least one thread waiting to add() can be unblocked, wake them up. if self._ready_to_unblock(): _LOGGER.debug("Notifying threads waiting to add messages to flow.") self._has_capacity.notify_all() def _distribute_available_capacity(self) -> None: """Distribute available capacity among the waiting threads in FIFO order. The method assumes that the caller has obtained ``_operational_lock``. """ available_slots = ( self._settings.message_limit - self._message_count - self._reserved_slots ) available_bytes = ( self._settings.byte_limit - self._total_bytes - self._reserved_bytes ) for reservation in self._waiting.values(): if available_slots <= 0 and available_bytes <= 0: break # Santa is now empty-handed, better luck next time. # Distribute any free slots. if available_slots > 0 and not reservation.has_slot: reservation.has_slot = True self._reserved_slots += 1 available_slots -= 1 # Distribute any free bytes. if available_bytes <= 0: continue bytes_still_needed = reservation.bytes_needed - reservation.bytes_reserved if bytes_still_needed < 0: # Sanity check for any internal inconsistencies. msg = "Too many bytes reserved: {} / {}".format( reservation.bytes_reserved, reservation.bytes_needed ) warnings.warn(msg, category=RuntimeWarning) bytes_still_needed = 0 can_give = min(bytes_still_needed, available_bytes) reservation.bytes_reserved += can_give self._reserved_bytes += can_give available_bytes -= can_give def _ready_to_unblock(self) -> bool: """Determine if any of the threads waiting to add a message can proceed. The method assumes that the caller has obtained ``_operational_lock``. """ if self._waiting: # It's enough to only check the head of the queue, because FIFO # distribution of any free capacity. first_reservation = next(iter(self._waiting.values())) return ( first_reservation.bytes_reserved >= first_reservation.bytes_needed and first_reservation.has_slot ) return False def _would_overflow(self, message: MessageType) -> bool: """Determine if accepting a message would exceed flow control limits. The method assumes that the caller has obtained ``_operational_lock``. Args: message: The message entering the flow control. """ reservation = self._waiting.get(threading.current_thread()) if reservation: enough_reserved = reservation.bytes_reserved >= reservation.bytes_needed has_slot = reservation.has_slot else: enough_reserved = False has_slot = False bytes_taken = self._total_bytes + self._reserved_bytes + message._pb.ByteSize() size_overflow = bytes_taken > self._settings.byte_limit and not enough_reserved msg_count_overflow = not has_slot and ( (self._message_count + self._reserved_slots + 1) > self._settings.message_limit ) return size_overflow or msg_count_overflow def _load_info( self, message_count: Optional[int] = None, total_bytes: Optional[int] = None ) -> str: """Return the current flow control load information. The caller can optionally adjust some of the values to fit its reporting needs. The method assumes that the caller has obtained ``_operational_lock``. Args: message_count: The value to override the current message count with. total_bytes: The value to override the current total bytes with. """ if message_count is None: message_count = self._message_count if total_bytes is None: total_bytes = self._total_bytes return ( f"messages: {message_count} / {self._settings.message_limit} " f"(reserved: {self._reserved_slots}), " f"bytes: {total_bytes} / {self._settings.byte_limit} " f"(reserved: {self._reserved_bytes})" )	11.82 KB	2025-05-25 12:26
📄 futures.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 typing from typing import Any, Callable, Union from google.cloud.pubsub_v1 import futures if typing.TYPE_CHECKING: # pragma: NO COVER from google.cloud import pubsub_v1 class Future(futures.Future): """This future object is returned from asychronous Pub/Sub publishing 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) -> str: """Return the message ID or raise an exception. This blocks until the message has been published successfully and returns the message ID unless an exception is raised. Args: timeout: The number of seconds before this call times out and raises TimeoutError. Returns: The message ID. Raises: concurrent.futures.TimeoutError: If the request times out. Exception: For undefined exceptions in the underlying call execution. """ return super().result(timeout=timeout) # This exists to make the type checkers happy. def add_done_callback( self, callback: Callable[["pubsub_v1.publisher.futures.Future"], Any] ) -> None: """Attach a callable that will be called when the future finishes. Args: callback: A callable that will be called with this future as its only argument when the future completes or is cancelled. The callable will always be called by a thread in the same process in which it was added. If the future has already completed or been cancelled then the callable will be called immediately. These callables are called in the order that they were added. """ return super().add_done_callback(callback) # type: ignore	2.84 KB	2025-05-25 12:26

Terminal