O 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$jRZ)ejTrddl+m,Z,ddlm-Z-ddl.m/Z/ddl0m1Z2ejfe4Z5e%jljoZ8eejrejtfZ;Gdde(jxZ=y))absolute_importN)AnyDictOptionalSequenceTupleTypeUnion)gapic_v1)AnonymousCredentials)service_account)types) exceptions)futures)thread)ordered_sequencer)unordered_sequencer)FlowController) gapic_version)client) pubsub_v1)_batch) OptionalRetry)pubsubceZdZdZ d'deej efdeejefde ffd Z e d(de deej efde ddfdZ e Zede fd Zefd Zd e d e defd Zd e d e ddfdZd)fd Zdej.j0ej.j0fd e ded e dddddeee fddfdZd*dZd*dZd*dZd*dZd*dZd*dZ d+d e d d!d e ddfd"Z!d#e"ddfd$Z# d+d e d%ed e ddfd&Z$xZ%S),ClientaA 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 t|tjust|dk(sJdt|tjust|dk(sJdt j jdr1dt j jdi|d<t|d<tj||_ |jd|_ t|0d i||jj|_t j"|_tj||_|j$j)|_i|_d|_d|_t3|jj4|_y) NrzBbatch_settings must be of type BatchSettings or an empty sequence.zHpublisher_options must be of type PublisherOptions or an empty sequence.PUBSUB_EMULATOR_HOST api_endpointclient_options credentialsF)typer BatchSettingslenPublisherOptionsosenvirongetr r_enable_message_orderingsuper__init__ _transport_host_targetrBatch _batch_classr make_lock _batch_lock _sequencers _is_stopped_commit_threadr flow_control_flow_controller)selfrrr __class__s :lib/third_party/google/cloud/pubsub_v1/publisher/client.pyr/zClient.__init__nsk  E$7$7 73~;NRS;S P O P S " #u'='= =$%* V V V + ::>>0 1 /E F(F# $%9$:F= !"'!7!79J!K(,(>(>q(A% "6",, "LL#11>B ,,668AC :>!/t/E/E/R/R Sfilenamereturnc \tjj|}||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)clsr@rrr$s r>rDz Client.from_service_account_files4*&11KKHU +}>,V,,r?c|jS)zeReturn the target (where the API is). Returns: The location of the API. )r2r<s r>targetz Client.targets||r?cRd}tj|ttS)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.)r<msgr=s r>apiz Client.apis% ,  c$67wr?topic ordering_keyc||f}|jj|}|B|dk(rtj||}nt j |||}||j|<|S)zdGet an existing sequencer or create a new one given the (topic, ordering_key) pair. )r7r,rUnorderedSequencerrOrderedSequencerr<rPrQ sequencer_key sequencers r>_get_or_create_sequencerzClient._get_or_create_sequencerss - $$((7  r!/BB4O ->>% /8D  ] +r?Nc0|j5|jr td|js t d||f}|j j |}|tjdn|jdddy#1swYyxYw)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.) r6r8 RuntimeErrorr- ValueErrorr7r,_LOGGERdebugunpauserVs r>resume_publishzClient.resume_publishs   "#RSS00 & #L1M((,,];I  ( !!#%  s A6B  Bc"t||i|S)z#Call the GAPIC public API directly.)r.publish)r<argsrr=s r>_gapic_publishzClient._gapic_publish sw///r?rSdataretryrtimeoutztypes.OptionalTimeoutattrsz"pubsub_v1.publisher.futures.Futurec 6t|ts tdjs|dk7r t dt j |j D]E\}}t|trt|tr|jd||<j@}|jCd }n|jCd }jE||}|j?|| } | jG| jI| cdddS#tj $r1} t#j$} | j'| | cYd} ~ Sd} ~ wwxYw#1swYyxYw) 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.rSzSCannot 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.)rerQ attributesNc<jjyN)r;release)futuremessager<s r>on_publish_donez'Client.publish..on_publish_done|s  ! ! ) )' 2r?z&Cannot publish on a stopped publisher.gA)rfrg)% isinstancebytes TypeErrorr-r\copyitemsstrdecode_raw_proto_pubbsub_message gapic_types PubsubMessagewrapr;addrFlowControlLimitErrorrFuture set_exceptionr methodDEFAULTrrfrgr6r8r[r0_wrapped_methodsrb_retry with_deadlinerYadd_done_callback!_ensure_commit_timer_runs_no_lock)r<rPrerQrfrgrhkv vanilla_pbexcrnrp transport base_retryrXros` @r>rbzClient.publish sGN$&O ,,1C+  IIe$**,DAq!S!!U#88G,a+  -0LU ++00<   ! ! % %g .  3 HOO++ +**00E hoo-- -,,44G   "#KLL ,,HOO333 $I!*!;!;I$ensure_cleanup_and_commit_timer_runsz+Client.ensure_cleanup_and_commit_timer_runss%     2 2 4  s'0c|js4|jjtdkr|j yyy)zEnsure a commit timer thread is running, without taking _batch_lock. _batch_lock must be held before calling this method. infN)r9r max_latencyfloat_start_commit_threadrGs r>rz(Client._ensure_commit_timer_runs_no_locks9 ""t':':'F'Fu'U  % % '(V"r?ctjd|jd|_|jj y)z>Start a new thread to actually wait and commit the sequencers.zThread-PubSubBatchCommitterT)namerHdaemonN) threadingThread_wait_and_commit_sequencersr9startrGs r>rzClient._start_commit_threads: (...33  !!#r?ctj|jjtj d|j 5|jr dddy|jd|_ dddy#1swYyxYw)z;Wait up to the batching timeout, and commit all sequencers.zCommit thread is waking upN) timesleeprrr]r^r6r8_commit_sequencersr9rGs r>rz"Client._wait_and_commit_sequencerssh 4&&223 23      # # %"&D    s B"BB c|jjDcgc]\}}|jr|}}}|D]}|j|=|jjD]}|j ycc}}w)z1Clean up finished sequencers and commit the rest.N)r7ru is_finishedvaluescommit)r<keyrXfinished_sequencer_keysrWs r>rzClient._commit_sequencerss#'"2"2"8"8":# ":Y$$& ": # 5M  /5))002I    3# sBc|j5|jr tdd|_|jj D]}|j  dddy#1swYyxYw)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)r6r8r[r7rstop)r<rXs r>rz Client.stopsY&  "#MNN#D !--446  7   s AA%%A.batchz_batch.thread.BatchcJ|j||}|j|yrl)rY _set_batch)r<rPrrQrXs r>rzClient._set_batchs$11%F U#r? batch_classc||_yrl)r4)r<rs r>_set_batch_classzClient._set_batch_classs 'r?rXc*||f}||j|<yrl)r7)r<rPrXrQrWs r>_set_sequencerzClient._set_sequencers - *3'r?)r%r%)r%)rAzpubsub_types.PublishResponse)rAN)rS)&__name__ __module__ __qualname____doc__r rr'rr)rr/ classmethodrvrDfrom_service_account_jsonpropertyrHrO SequencerTyperYr`rdr rrrrrbrrrrrrrr rr __classcell__)r=s@r>rr<s/f@BEG,Te118;<,T!!7!7!AB,T ,T\@B--e118;<- -  --0!: &c""$C"$s"$t"$H0!)!8!8+3??+B+B QQQ Q  Q ) Qucz"Q .Qf5( $ ' !<MO$$!6$FI$ $(D(T( IK44%24BE4 4r?r)> __future__rrtloggingr*rrtypingrrrrrr r rKgoogle.api_corer google.auth.credentialsr google.oauth2r google.cloud.pubsub_v1r google.cloud.pubsub_v1.publisherrr'google.cloud.pubsub_v1.publisher._batchr+google.cloud.pubsub_v1.publisher._sequencerrr0google.cloud.pubsub_v1.publisher.flow_controllerrgoogle.pubsub_v1rpackage_versionry#google.pubsub_v1.services.publisherrpublisher_client __version__ TYPE_CHECKING google.cloudrr*google.pubsub_v1.services.publisher.clientrgoogle.pubsub_v1.typesr pubsub_types getLoggerrr]rzpbrxrUrTrPublisherClientrr%r?r>rs'   DDD$8)(74:IKK=1J))  &7H= '  H %)6699;&&(;(N(NN F4  - -F4r?