U
    d-                     @   s   d Z ddlZddlmZ ddlmZ ddlmZ ddlm	Z
 ddlmZmZ ddlmZ dd	lmZ dd
lmZmZmZmZ ddlmZ ddlmZ dZeeZdadZG dd de Z!dd Z"edd Z#dd Z$dd Z%G dd dZ&dS )zEvent loop implementation.    N)contextmanager)Empty)sleep)GeneratorType)Thenablepromise)
get_logger)fileno)ERRREADWRITEpoll)cached_property   )Timer)Hubget_event_loopset_event_loopz<Received unknown event %r for fd %r, please contact support!c                   @   s   e Zd ZdZdS )StopzStops the event loop.N)__name__
__module____qualname____doc__ r   r   :/tmp/pip-unpacked-wheel-hqfrjlvz/kombu/asynchronous/hub.pyr      s   r   c                   C   s
   t  d S N)r   r   r   r   r   _raise_stop_error    s    r   c                  o   s
   d V  d S r   r   )argskwargsr   r   r   _dummy_context$   s    r   c                   C   s   t S )zGet current event loop object._current_loopr   r   r   r   r   )   s    r   c                 C   s   | a | S )z"Set the current event loop object.r    )loopr   r   r   r   .   s    r   c                
   @   sF  e Zd ZdZeZeZeZdZdDddZe	dd Z
e
jdd Z
dd	 Zd
d Zdd Zdd Zdd ZdEddZdd ZdFddZdd Zdd Zd d! Zd"d# Zd$d% Zd&d' Zd(d) Zd*d+ Zd,d- Zd.d/ Zd0d1 Zd2d3 Zd4d5 Z d6d7 Z!d8d9 Z"e#e$e%e&e'e(e)eeef
d:d;Z*d<d= Z+d>d? Z,e-d@dA Z.e	dBdC Z/dS )Gr   zpEvent loop object.

    Arguments:
        timer (kombu.asynchronous.Timer): Specify custom timer instance.
    Nc                 C   sd   |d k	r|nt  | _i | _i | _t | _t | _t | _d| _d | _	t | _
d | _d| _|   d S )NFr   )r   timerreaderswritersseton_tickon_close_ready_running_loopconsolidateconsolidate_callbackpropagate_errors_create_poller)selfr#   r   r   r   __init__I   s    	zHub.__init__c                 C   s   | j s|   | j S r   )_pollerr/   r0   r   r   r   pollerc   s    z
Hub.pollerc                 C   s
   || _ d S r   )r2   )r0   valuer   r   r   r4   i   s    c                 C   s   |    |   d S r   )closer/   r3   r   r   r   resetm   s    z	Hub.resetc                 C   s    t  | _| jj| _| jj| _d S r   )r   r2   register_register_fd
unregister_unregister_fdr3   r   r   r   r/   q   s    
zHub._create_pollerc                 C   s*   | j d k	r&| j   d | _ d | _d | _d S r   )r2   r6   r9   r;   r3   r   r   r   _close_pollerv   s
    

zHub._close_pollerc                 C   s   |  t d S r   )	call_soonr   r3   r   r   r   stop}   s    zHub.stopc                 C   s   d t| t| jt| jS )Nz<Hub@{:#x}: R:{} W:{}>)formatidlenr$   r%   r3   r   r   r   __repr__   s
      zHub.__repr__r   
   r   c           
      C   s   | j }d }|r|jrt|D ]}t| j\}}|d kr: qz
|  W q |k
rZ    Y q ttfk
rr    Y q tk
r }	 z"|	jtj	kr t
jd|	dd W 5 d }	~	X Y q tk
r }	 zt
jd|	dd W 5 d }	~	X Y qX qt|p||S )NzError in timer: %rr   exc_info)r#   _queuerangenext	schedulerMemoryErrorAssertionErrorOSErrorerrnoZENOMEMloggererror	Exceptionmin)
r0   Z	min_delayZ	max_delayZ
max_timers	propagater#   delayientryexcr   r   r   fire_timers   s(    

 $zHub.fire_timersc              	   C   s    z| | W 5 |  | X d S r   )_discard_unregisterr0   fdr   r   r   _remove_from_loop   s    zHub._remove_from_loopFc                 C   sz   t |}z| j|| W n  tk
r:   | |  Y n<X |t@ rJ| jn| j}|rj| j	| d ||< n||f||< d S r   )
r	   r4   r8   
ValueErrorr\   r   r$   r%   r,   add)r0   r[   callbackflagsr   r,   destr   r   r   r^      s    

zHub.addc                 C   s   t |}| | d S r   )r	   r\   rZ   r   r   r   remove   s    z
Hub.removec                 C   s@   d| _ z,z|   W q tk
r,   Y q0Y qX qW 5 d| _ X d S )NTF)r*   run_oncer   r3   r   r   r   run_forever   s    zHub.run_foreverc                 C   s.   zt | j W n tk
r(   d | _Y nX d S r   )rH   r"   StopIterationr+   r3   r   r   r   rc      s    zHub.run_oncec                 G   s$   t |tst||}| j| |S r   )
isinstancer   r   r)   r^   )r0   r_   r   r   r   r   r=      s    

zHub.call_soonc                 G   s   | j |||S r   )r#   Z
call_afterr0   rS   r_   r   r   r   r   
call_later   s    zHub.call_laterc                 G   s   | j |||S r   )r#   call_at)r0   whenr_   r   r   r   r   ri      s    zHub.call_atc                 G   s   | j |||S r   )r#   call_repeatedlyrg   r   r   r   rk      s    zHub.call_repeatedlyc                 G   s   |  ||ttB |S r   )r^   r   r
   r0   Zfdsr_   r   r   r   r   
add_reader   s    zHub.add_readerc                 G   s   |  ||t|S r   )r^   r   rl   r   r   r   
add_writer   s    zHub.add_writerc                 C   sH   || j k}| j |}z| | W 5 |rB|\}}| ||t| X d S r   )r%   getr^   r   r\   )r0   r[   writableZon_writecbr   r   r   r   remove_reader   s    
zHub.remove_readerc                 C   sL   || j k}| j |}z| | W 5 |rF|\}}| ||ttB | X d S r   )r$   ro   r^   r   r
   r\   )r0   r[   readableZon_readrq   r   r   r   r   remove_writer   s    
zHub.remove_writerc              
   C   s0   z| j | W n tttfk
r*   Y nX d S r   )r4   r:   AttributeErrorKeyErrorrL   rZ   r   r   r   rY      s    zHub._unregisterc                    s    fdd j D   j    fdd jD   j   j      jD ]}|  qTt j}t  _|D ]
}|  qxd S )Nc                    s   g | ]}  |qS r   rY   .0r[   r3   r   r   
<listcomp>   s     zHub.close.<locals>.<listcomp>c                    s   g | ]}  |qS r   rw   rx   r3   r   r   rz      s     )	r$   clearr%   r,   r<   r(   listr)   r&   )r0   r   r_   Ztodositemr   r3   r   r6      s    





z	Hub.closec                 C   s4   t |}| j|d  | j|d  | j| d S r   )r	   r$   popr%   r,   discardrZ   r   r   r   rX   	  s    zHub._discardc                 C   s   t jd||dd d S )Nz Callback %r raised exception: %rr   rD   )rN   rO   )r0   r_   rV   r   r   r   on_callback_error  s       zHub.on_callback_errorc           !      c   s  | j | j }}| jj}| j}| j}| jj}| j}| j	}| j
}| j}| j}t | _|D ]
}|  qT|D ]}|rd|  qd|r||dnd}|s|rg }z||}W n tk
r   Y d S X |pdD ]\}}d}||kr||d kr|| qd  }}||@ r:z|| \}}W n$ |k
r6   | | Y qY nX nd||	@ r|z|| \}}W n$ |k
rx   | | Y qY nX n"||
@ rd}ntt|| d}|rz||p||\}}W n tk
r   Y nX |d kr| | qt||rzz|| W nn tk
rD }  z| jtjkr, || W 5 d } ~ X Y n4 |k
rX   Y n  tk
rv   ||  Y nX qz||  W q |k
r   Y qX q|r|| n|||d d V  qBd S )N)rR   r   r   FTg?)r$   r%   r4   r   rW   rb   r#   rF   r,   r-   r'   r.   r)   r&   r]   ro   appendrr   rt   rN   infoW_UNKNOWN_EVENT	TypeErrorrf   rL   rM   EBADFrP   )!r0   	generatorr   rQ   rH   r   re   rv   r   r   r
   r$   r%   r   rW   Z
hub_removeZ	scheduledr,   r-   r'   rR   todoZtick_callbackr}   Zpoll_timeoutZto_consolidateeventsr[   eventZgeneral_errorrq   ZcbargsrV   r   r   r   create_loop  s    











zHub.create_loopc                 C   s   ddl m} || S )Nr   )repr_active)debugr   )r0   r   r   r   r   r   t  s    zHub.repr_activec                 C   s   ddl m} || |pg S )Nr   )repr_events)r   r   )r0   r   r   r   r   r   r   x  s    zHub.repr_eventsc                 C   s
   t | jS r   )iterr#   r3   r   r   r   rI   |  s    zHub.schedulerc                 C   s   | j d kr|  | _ | j S r   )r+   r   r3   r   r   r   r"     s    

zHub.loop)N)r   rC   rC   r   )r   F)0r   r   r   r   r   r   r
   r(   r1   propertyr4   setterr7   r/   r<   r>   rB   rW   r\   r^   rb   rd   rc   r=   rh   ri   rk   rm   rn   rr   rt   rY   r6   rX   r   r   r   rQ   rH   r   re   rv   r   r   r   r   rI   r"   r   r   r   r   r   5   sh   


  



       
`
r   )'r   rM   
contextlibr   queuer   timer   typesr   r   Zviner   r   Z	kombu.logr   Zkombu.utils.compatr	   Zkombu.utils.eventior
   r   r   r   Zkombu.utils.objectsr   r#   r   __all__r   rN   r!   r   BaseExceptionr   r   r   r   r   r   r   r   r   r   <module>   s,   
