
                             S SK rS SKrS SKJr  S SKJrJr  S SKJ	r	  S SK
Jr  S SKJr  S SKJr  S SKJrJr  S S	KJr  S S
KJr   " S S\\5      rg)    N)ThreadPoolExecutor)ContextManagerOptional)GoogleAPICallError)partial)wait_ignore_errors)ManagedEventLoop)StreamingPullManagerCloseCallback)AsyncSingleSubscriber)MessageCallbackc                      \ rS rSr% \\S'   \\S'   \\S'   \\S'   \	R                  R                  \S'   \R                  \S'   \\   \S'   \\   \S	'   \\S
'   S\S\S\4S jrS\4S jrS rS\4S jrS rS rS rSrg)SubscriberImpl&   _underlying	_callback_unowned_executor_event_loop_poller_future_close_lock_failure_close_callback_closed
underlyingcallbackunowned_executorc                     Xl         X l        X0l        [        S5      U l        [
        R                  " 5       U l        S U l        S U l	        SU l
        g )NSubscriberLoopThreadF)r   r   r   r	   r   	threadingLockr   r   r   r   )selfr   r   r   s       Olib/third_party/google/cloud/pubsublite/cloudpubsub/internal/subscriber_impl.py__init__SubscriberImpl.__init__3   sH     &!!1+,BC$>>+#    close_callbackc                 z    U R                      U R                  b   eXl        SSS5        g! , (       d  f       g= f)z
A close callback must be set exactly once by the StreamingPullFuture managing this subscriber.

This two-phase init model is made necessary by the requirements of StreamingPullFuture.
N)r   r   )r!   r&   s     r"   add_close_callback!SubscriberImpl.add_close_callbackB   s1     ''///#1  s   ,
:c                     U R                      U R                  (       a
   S S S 5        g SU l        S S S 5        U R                  S S S 5        g ! , (       d  f       N"= f)NT)r   r   __exit__r!   s    r"   closeSubscriberImpl.closeL   sF    ||   DL  	dD$'	 s   AA
Aerrorc                 0    Xl         U R                  5         g N)r   r-   )r!   r/   s     r"   _failSubscriberImpl._failS   s    

r%   c                 2  #      U R                   R                  5       I S h  vN nU R                  R                  U R                  U5        MJ   N,! [
         a9  nU R                  R                  [        U R                  U5      5         S nAg S nAff = f7fr1   )	r   readr   mapr   r   submitr   r2   )r!   batches      r"   _pollerSubscriberImpl._pollerW   sz     	B"..3355&&**4>>5A 5! 	B""))'$**a*@AA	Bs2   BA A-A 
B/B
BBBc                 6   U R                   c   eU R                  R                  5         U R                  R                  U R                  R                  5       5      R                  5         U R                  R                  U R                  5       5      U l        U $ r1   )	r   r   	__enter__r7   r   
__aenter__resultr:   r   r,   s    r"   r=   SubscriberImpl.__enter___   sv    ##///""$ 0 0 ; ; =>EEG"..55dllnEr%   c           	         U R                   R                  5          U R                   R                  5         U R                  R	                  [        U R                  R                  XU5      5      5      R                  5         U R                  R                  XU5        U R                  c   eU R                  X R                  5        g !    N= fr1   )r   cancelr?   r   r7   r   r   	__aexit__r+   r   r   )r!   exc_type	exc_value	tracebacks       r"   r+   SubscriberImpl.__exit__f   s    ""$	&&( 	  **8	J	
 &(!!(yA##///T==1	s   C	 	C)	r   r   r   r   r   r   r   r   r   N)__name__
__module____qualname____firstlineno__r   __annotations__r   r   r	   
concurrentfuturesFuturer   r    r   r   r   boolr#   r(   r-   r2   r:   r=   r+   __static_attributes__ r%   r"   r   r   &   s    &&))!!&&---)**m,,M) " -	2 2(- B2r%   r   )concurrent.futuresrM   r   concurrent.futures.threadr   typingr   r   google.api_core.exceptionsr   	functoolsr   6google.cloud.pubsublite.internal.wait_ignore_cancelledr   ?google.cloud.pubsublite.cloudpubsub.internal.managed_event_loopr	   Cgoogle.cloud.pubsublite.cloudpubsub.internal.streaming_pull_managerr
   r   >google.cloud.pubsublite.cloudpubsub.internal.single_subscriberr   ?google.cloud.pubsublite.cloudpubsub.subscriber_client_interfacer   r   rR   r%   r"   <module>r]      sA      8 + 9  U
M2^%9 M2r%   