a !fP@s|ddlmZddlZddlZddlZddlZddlZddlZddlmZm Z m Z m Z m Z m Z mZddlZddlmZddlmZddlmZddlmZddlmZdd lmZdd lmZdd lmZdd lmZdd l m!Z!ddl"m#Z$ddl"mZ%ddl&m'Z(e$j)Z)ej*rBddl+m,Z,ddlm-Z-ddl.m/Z/ddl0m1Z2e3e4Z5e%j67Z8eej9ej:fZ;Gddde(j<Z=dS))absolute_importN)AnyDictOptionalSequenceTupleTypeUnion)gapic_v1)AnonymousCredentials)service_account)types) exceptions)futures)thread)ordered_sequencer)unordered_sequencer)FlowController) gapic_version)client) pubsub_v1)_batch) OptionalRetry)pubsubc seZdZdZd4eejefeejefe dfdd Z e d5e eejefe ddddZ e Zee d d d Zefd d Ze e edddZe e ddddZdd fdd Zdejjejjfe ee ddeee fddddZdd ddZdd d d!Zdd d"d#Zdd d$d%Zdd d&d'Zdd d(d)Z d6e d*e dd+d,d-Z!e"dd.d/d0Z#d7e ee dd1d2d3Z$Z%S)8ClientaA 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 } ) )batch_settingspublisher_optionskwargsc st|tjus"t|dks"Jdt|tjusDt|dksDJdtjdrndtjdi|d<t|d<tj||_ |j d|_ t j fi||j j|_tj|_tj||_|j|_i|_d|_d|_t|j j|_dS) NrzBbatch_settings must be of type BatchSettings or an empty sequence.zHpublisher_options must be of type PublisherOptions or an empty sequence.ZPUBSUB_EMULATOR_HOSTZ api_endpointZclient_options credentialsF)typer BatchSettingslenPublisherOptionsosenvirongetr r_enable_message_orderingsuper__init__ _transportZ_host_targetrZBatch _batch_classrZ make_lock _batch_lock _sequencers _is_stopped_commit_threadrZ flow_control_flow_controller)selfrrr __class__rh/var/www/html/python-backend/venv/lib/python3.9/site-packages/google/cloud/pubsub_v1/publisher/client.pyr)ns6            zClient.__init__)filenamerrreturncKs$tj|}||d<||fi|S)aCreates 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. r)r Credentialsfrom_service_account_file)clsr6rrrrrr5r9s z Client.from_service_account_file)r7cCs|jS)zeReturn the target (where the API is). Returns: The location of the API. )r+r2rrr5targetsz Client.targetcsd}tj|tdtS)aThe 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. zThe "api" property only exists for backward compatibility, access its attributes directly thorugh the client instance (e.g. "client.foo" instead of "client.api.foo").)category)warningswarnDeprecationWarningr()r2msgr3rr5apis z Client.api)topic ordering_keyr7cCsN||f}|j|}|durJ|dkr2t||}nt|||}||j|<|S)zdGet an existing sequencer or create a new one given the (topic, ordering_key) pair. N)r.r&rUnorderedSequencerrOrderedSequencerr2rCrD sequencer_key sequencerrrr5_get_or_create_sequencers  zClient._get_or_create_sequencerNcCsv|j\|jrtd|js$td||f}|j|}|durLtdn| Wdn1sh0YdS)a*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. z-Cannot resume publish on a stopped publisher.zICannot resume publish on a topic/ordering key if ordering is not enabled.NzCError: The topic/ordering key combination has not been seen before.) r-r/ RuntimeErrorr' ValueErrorr.r&_LOGGERdebugZunpauserHrrr5resume_publishs zClient.resume_publishzpubsub_types.PublishResponsecstj|i|S)z#Call the GAPIC public API directly.)r(publish)r2argsrr3rr5_gapic_publish szClient._gapic_publishrErztypes.OptionalTimeoutz"pubsub_v1.publisher.futures.Future)rCdatarDretrytimeoutattrsr7c st|tstdjs(|dkr(tdt|D]6\}}t|trJq6t|trd|d||<q6tdq6t |||d} t j | zj Wn<tjy} z"t} | | | WYd} ~ Sd} ~ 00fdd } |tjjurjj}|tjjurjj}jjr(td jrz|tjjurbj} | j| jj }|!d }d }n|durz|!d }d }"||}|j||d } | #| $| WdS1s0YdS) aN 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. z=Data being published to Pub/Sub must be sent as a bytestring.rEzSCannot publish a message with an ordering key when message ordering is not enabled.zutf-8zGAll attributes being published to Pub/Sub must be sent as text strings.)rTrD attributesNcsjdSN)r1release)futuremessager2rr5on_publish_done|sz'Client.publish..on_publish_donez&Cannot publish on a stopped publisher.gA)rUrV)% isinstancebytes TypeErrorr'rMcopyitemsstrdecode_raw_proto_pubbsub_message gapic_types PubsubMessagewrapr1addrZFlowControlLimitErrorrFuture set_exceptionr methodDEFAULTrrUrVr-r/rLr*Z_wrapped_methodsrQZ_retryZ with_deadlinerKadd_done_callback!_ensure_commit_timer_runs_no_lock)r2rCrTrDrUrVrWkvZ vanilla_pbexcr[r^ transportZ base_retryrJrr\r5rQ sbG          zClient.publishcCs2|j|Wdn1s$0YdS)zEnsure a cleanup/commit timer thread is running. If a cleanup/commit timer thread is already running, this does nothing. N)r-rpr;rrr5$ensure_cleanup_and_commit_timer_runssz+Client.ensure_cleanup_and_commit_timer_runscCs"|js|jjtdkr|dS)zEnsure a commit timer thread is running, without taking _batch_lock. _batch_lock must be held before calling this method. infN)r0r max_latencyfloat_start_commit_threadr;rrr5rpsz(Client._ensure_commit_timer_runs_no_lockcCs"tjd|jdd|_|jdS)z>Start a new thread to actually wait and commit the sequencers.zThread-PubSubBatchCommitterT)namer<daemonN) threadingThread_wait_and_commit_sequencersr0startr;rrr5rys zClient._start_commit_threadcCsft|jjtd|j4|jr6WddS|d|_ Wdn1sX0YdS)z;Wait up to the batching timeout, and commit all sequencers.zCommit thread is waking upN) timesleeprrwrNrOr-r/_commit_sequencersr0r;rrr5r~s z"Client._wait_and_commit_sequencerscCsBdd|jD}|D] }|j|=q|jD] }|q0dS)z1Clean up finished sequencers and commit the rest.cSsg|]\}}|r|qSr) is_finished).0keyrJrrr5 sz-Client._commit_sequencers..N)r.rcvaluescommit)r2Zfinished_sequencer_keysrIrJrrr5rs zClient._commit_sequencerscCsV|j<|jrtdd|_|jD] }|q&Wdn1sH0YdS)aImmediately 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. z(Cannot stop a publisher already stopped.TN)r-r/rLr.rstop)r2rJrrr5rs z Client.stopz_batch.thread.Batch)rCbatchrDr7cCs|||}||dSrY)rK _set_batch)r2rCrrDrJrrr5rs zClient._set_batch) batch_classr7cCs ||_dSrY)r,)r2rrrr5_set_batch_classszClient._set_batch_class)rCrJrDr7cCs||f}||j|<dSrY)r.)r2rCrJrDrIrrr5_set_sequencerszClient._set_sequencer)rr)r)rE)rE)&__name__ __module__ __qualname____doc__r r r!rr#rr) classmethodrdr9Zfrom_service_account_jsonpropertyr<rB SequencerTyperKrPrSr rmrnr`rQrurpryr~rrrrrr __classcell__rrr3r5r<sl3  . $       r)> __future__rrbloggingr$r|rtypingrrrrrrr r>Zgoogle.api_corer Zgoogle.auth.credentialsr Z google.oauth2r Zgoogle.cloud.pubsub_v1r Z google.cloud.pubsub_v1.publisherrrZ'google.cloud.pubsub_v1.publisher._batchrZ+google.cloud.pubsub_v1.publisher._sequencerrrZ0google.cloud.pubsub_v1.publisher.flow_controllerrZgoogle.pubsub_v1rZpackage_versionrgZ#google.pubsub_v1.services.publisherrZpublisher_client __version__ TYPE_CHECKINGZ google.cloudrrZ*google.pubsub_v1.services.publisher.clientrZgoogle.pubsub_v1.typesrZ pubsub_types getLoggerrrNrhZpbrfrGrFrZPublisherClientrrrrr5sB $