File: //usr/lib64/python3.6/site-packages/zmq/__pycache__/_future.cpython-36.opt-1.pyc
3
VS�_cH � @ sn d Z ddlmZmZ ddlmZ ddlmZmZm Z ddlZ
edd�Z G d d � d e
j �Z
G d
d� de
j�ZdS )z%Future-returning APIs for coroutines.� )�
namedtuple�deque)�chain)�EVENTS�POLLOUT�POLLINN� _FutureEvent�future�kind�kwargs�msgc s" e Zd ZdZd� fdd� Z� ZS )� _AsyncPollerz:Poller that returns a Future on poll, instead of blocking.� c s� �j � � |dkr`ytt��jd�}W n, tk
rP } z� j|� W Y dd}~X n X � j|� � S �j� ��j � �g ��fdd�}�j���fdd�� x��j D ]�\}}t
|t j �r�t
|�j
�sʈj
j|�}|t j@ r�|jd�d� |t j@ r�|jd�d� q��j|� d}|t j@ �r |�jO }|t j@ �r6|�jO }�j�|||� q�W �� ��fd d
�}�j|� |dk �r�|dk�r��fd d �} �jd
| | ����fdd�}
� j|
� �fdd�} � j| � � S )z Return a Future for a poll eventr Nc s � j � s� jd � d S )N)�done�
set_result)�args)�watcher� � /usr/lib64/python3.6/_future.py�wake_raw- s z#_AsyncPoller.poll.<locals>.wake_rawc s �j � f��� S )N)Z_unwatch_raw_sockets)�f)�loop� raw_sockets�selfr r �<lambda>1 s z#_AsyncPoller.poll.<locals>.<lambda>�poll)r c s� �j � r d S �j� r:y �j� W n tk
r4 Y nX d S �j� rR�j�j� � nLytt��jd�}W n, t k
r� } z�j|� W Y d d }~X n X �j
|� d S )Nr ) r � cancelled�cancel� RuntimeError� exception�
set_exception�superr
r � Exceptionr )r �result�e)� __class__r r r r r �
on_poll_readyE s z(_AsyncPoller.poll.<locals>.on_poll_readyc s � j � s� jd � d S )N)r r r )r r r �trigger_timeout\ s z*_AsyncPoller.poll.<locals>.trigger_timeoutg����MbP?c s" t �d�r�j� n
� j�� d S )Nr )�hasattrr Zremove_timeout)r )r �timeout_handler r �cancel_timeoutd s
z)_AsyncPoller.poll.<locals>.cancel_timeoutc s � j � s� j� d S )N)r r )r )r r r �cancel_watcherk s z)_AsyncPoller.poll.<locals>.cancel_watcher)�_Futurer! r
r r" r r �
_default_loop�add_done_callbackZsockets�
isinstance�_zmq�SocketZ
_socket_class� from_socketr �_add_recv_eventr �_add_send_event�append�_READZ_WRITEZ_watch_raw_socket�
call_later) r �timeoutr# r$ r �socket�maskZevtr& r' r* r+ )r% )r r r r r) r r r sP
z_AsyncPoller.poll���)r; )�__name__�
__module__� __qualname__�__doc__r �
__classcell__r r )r% r r
s r
c s6 e Zd ZdZdZdZdZeZdZ dZ
d7� fdd� Z e d8dd��Z
d9� fdd � Zejjje_� fd
d �Zejjje_d:dd�Zd;dd�Zd<dd�Zd=dd�Zdd� Zdejfdd�Zdd� Zd d � Ze d d � �Z d>d d!�Z d?d"d#�Z d$d%� Z d&d'� Z!d@d(d)�Z"dAd*d+�Z#d,d-� Z$d.d/� Z%d0d1� Z&d2d3� Z'd4d5� Z(� Z)S )B� _AsyncSocketNr r c s� t |tj�rd | }}n |jdd �}|d k rHtt| �j|jd� || _n&tt| �j||f|� tjj | j�| _|px| j
� | _ t � | _
t � | _d| _| jj| _| j� d S )N� _from_socket)�shadowr )r/ r0 r1 �popr! rA �__init__Z
underlying� _shadow_sockrC r- �io_loopr �
_recv_futures�
_send_futures�_stateZFD�_fd�_init_io_state)r �contextZ socket_typerG r r2 )r% r r rE � s
z_AsyncSocket.__init__c C s | ||d�S )z.Create an async socket from an existing Socket)rB rG r )�clsr9 rG r r r r2 � s z_AsyncSocket.from_socketc sn | j sXxHtt| j| j��D ]2}|jj� sy|jj� W q tk
rJ Y qX qW | j � t
t | �j |d� d S )N)�linger)
�closed�listr rH rI r r r r �_clear_io_stater! rA �close)r rO Zevent)r% r r rS � s
z_AsyncSocket.closec s& t t| �j|�}|tkr"| j|� |S )N)r! rA �getr �_schedule_remaining_events)r �keyr# )r% r r rT � s
z_AsyncSocket.getTFc C s | j dt|||d��S )z~Receive a complete multipart zmq message.
Returns a Future whose result will be a multipart message.
�recv_multipart)�flags�copy�track)r3 �dict)r rX rY rZ r r r rW � s z_AsyncSocket.recv_multipartc C s | j dt|||d��S )z�Receive a single zmq frame.
Returns a Future, whose result will be the received frame.
Recommend using recv_multipart instead.
�recv)rX rY rZ )r3 r[ )r rX rY rZ r r r r\ � s z_AsyncSocket.recvc K s( ||d< ||d< ||d< | j d||d�S )zqSend a complete multipart zmq message.
Returns a Future that resolves when sending is complete.
rX rY rZ �send_multipart)r r )r4 )r r rX rY rZ r r r r r] � s z_AsyncSocket.send_multipartc K s< ||d< ||d< ||d< |j t|||d�� | jd||d�S )z�Send a single zmq frame.
Returns a Future that resolves when sending is complete.
Recommend using send_multipart instead.
rX rY rZ )rX rY rZ �send)r r )�updater[ r4 )r r rX rY rZ r r r r r^ � s
z_AsyncSocket.sendc s> | j � � � ��fdd�}�j|� � �fdd�}� j|� � S )zDeserialize with Futuresc st � j � r dS �j� r$� j�j� � nL�j� }y �|�}W n, tk
rd } z� j|� W Y dd}~X n X � j|� dS )z+Chain result through serialization to recvdN)r r r r# r" r )�_ZbufZloadedr$ )r �load�recvdr r �_chain� s z)_AsyncSocket._deserialize.<locals>._chainc s �j � r dS � j� r �j� dS )z"Chain cancellation from f to recvdN)r r r )r` )r rb r r �
_chain_cancel� s z0_AsyncSocket._deserialize.<locals>._chain_cancel)r, r. )r rb ra rc rd r )r ra rb r � _deserialize� s
z_AsyncSocket._deserializec sf �j rtjtj���j� }|j�|� |j|�}�j� � � �fdd�}|j� rX||� n
|j |� � S )zSpoll the socket for events
returns a Future for the poll results.
c st � j � r d S | j� r:y � j� W n tk
r4 Y nX d S | j� rR� j| j� � n t| j� �}� j|j �d�� d S )Nr )
r r r r r r r[ r# r rT )r Zevts)r r r r �
unwrap_result s z(_AsyncSocket.poll.<locals>.unwrap_result)
rP r0 ZZMQErrorZENOTSUP�
_poller_class�registerr r, r r. )r r8 rX �pr rf r )r r r r � s
z_AsyncSocket.pollc s � fdd�}| j ||� dS )z'Add a timeout for a send or recv Futurec s � j � r d S � jtj� � d S )N)r r r0 �Againr )r r r �future_timeout s z1_AsyncSocket._add_timeout.<locals>.future_timeoutN)� _call_later)r r r8 rk r )r r � _add_timeout s z_AsyncSocket._add_timeoutc C s | j j||� dS )z�Schedule a function to be called later
Override for different IOLoop implementations
Tornado and asyncio happen to both have ioloop.call_later
with the same signature.
N)rG r7 )r Zdelay�callbackr r r rl # s z_AsyncSocket._call_laterc C s> x*t |�D ]\}\}}}}|| kr
P q
W dS |j|| � dS )z�Make sure that futures are removed from the event list when they resolve
Avoids delaying cleanup until the next send/recv event,
which may never come.
N)� enumerate�remove)r Z
event_listZf_idxr r
r r` r r r �_remove_finished_future- s
z$_AsyncSocket._remove_finished_futurec s� |p
� j � }|jd�r~|jdd�tj@ r~t� j|�}y|f |�}W n, tk
rn } z|j|� W Y dd}~X n X |j |� |S � j
j t |||dd�� |j
� fdd�� ttd�r҈ jj}|dkr҈ j||d � � jjt�t@ r� j� � j
r�� jt� |S )
z4Add a recv event, returning the corresponding Futurer\ rX r N)r c s � j | � j�S )N)rq rH )r )r r r r R s z._AsyncSocket._add_recv_event.<locals>.<lambda>ZRCVTIMEOg����MbP?)r, �
startswithrT r0 �DONTWAIT�getattrrF r" r r rH r5 r r. r( Zrcvtimeorm r r � _handle_recv�
_add_io_state) r r
r r r r\ �rr$ �
timeout_msr )r r r3 = s*
z _AsyncSocket._add_recv_eventc
sT |p
� j � }|d
kr� j r�|jdd�}|j� }|tjB |d< t� j|�}d} y||f|�}
W nh tjk
r� } z |tj@ r�|j | � nd} W Y dd} ~ X n6 t
k
r� } z|j | � W Y dd} ~ X n X |j |
� | r� j r� j
� |S � jjt||||d�� |j� fd d
�� ttd ��rF� jjtj�} | dk�rF� j|| d � � jt� |S )z4Add a send event, returning the corresponding Futurer^ r] rX r TFN)r r c s � j | � j�S )N)rq rI )r )r r r r � s z._AsyncSocket._add_send_event.<locals>.<lambda>�SNDTIMEOg����MbP?)r^ r] )r, rI rT rY r0 rs rt rF rj r r" r rH rU r5 r r. r( ry rm rv r )
r r
r r r r rX Z
nowait_kwargsr^ Z finish_earlyrw r$ rx r )r r r4 ` s<
z _AsyncSocket._add_send_eventc C s | j jt�t@ sdS d}x,| jrD| jj� \}}}}|j� r@d}qP qW | jsV| jt� |dkrbdS |dkrx|jd� dS |dkr�| j j }n |dkr�| j j
}n t d| ��|d t j
O < y|f |�}W n, tk
r� } z|j|� W Y dd}~X n X |j|� dS )zHandle recv eventsNr rW r\ z Unhandled recv event type: %rrX )rF rT r r rH �popleftr �_drop_io_stater rW r\ �
ValueErrorr0 rs r" r )r r r
r r` r\ r# r$ r r r ru � s4
z_AsyncSocket._handle_recvc C s | j jt�t@ sd S d }x,| jrD| jj� \}}}}|j� r@d }qP qW | jsV| jt� |d krbd S |dkrx|jd � d S |dkr�| j j }n |dkr�| j j
}n t d| ��|d t j
O < y||f|�}W n, tk
r� } z|j|� W Y d d }~X n X |j|� d S )Nr r] r^ z Unhandled send event type: %rrX )rF rT r r rI rz r r{ r r] r^ r| r0 rs r" r )r r r
r r r^ r# r$ r r r � _handle_send� s4
z_AsyncSocket._handle_sendc C s< | j jt�}|tj@ r | j� |tj@ r0| j� | j� dS )z(Dispatch IO events to _handle_recv, etc.N) rF rT r r0 r ru r r} rU )r �fd�eventsZ
zmq_eventsr r r �_handle_events� s
z_AsyncSocket._handle_eventsc C s> | j dkrdS |dkr"| jjt�}|| j @ r:| jd| j� dS )zkSchedule a call to handle_events next loop iteration
If there are still events to handle.
r N)rJ rF rT r rl r� )r r r r r rU � s
z'_AsyncSocket._schedule_remaining_eventsc C s* | j |kr| j |B }| _ | j| j � dS )zAdd io_state to poller.N)rJ �_update_handler)r �stater r r rv � s
z_AsyncSocket._add_io_statec C s( | j |@ r| j | @ | _ | j| j � dS )z&Stop poller from watching an io_state.N)rJ r� )r r� r r r r{ s
z_AsyncSocket._drop_io_statec C s | j � dS )zOUpdate IOLoop handler with state.
zmq FD is always read-only.
N)rU )r r� r r r r� s z _AsyncSocket._update_handlerc C s( | j j| j| j| j� | jd| j� dS )z#initialize the ioloop event handlerr N)rG Z add_handlerrF r� r6 rl )r r r r rL s z_AsyncSocket._init_io_statec C s$ | j }| j jr| j}| jj|� dS )zNunregister the ioloop event handler
called once during close
N)rF rP rK rG Zremove_handler)r r~ r r r rR s z _AsyncSocket._clear_io_stater; )Nr; N)N)N)r TF)r TF)r TF)r TF)NN)NNN)r r )N)*r< r= r> rH rI rJ rF r
rg rG rK rE � classmethodr2 rS r0 r1 r? rT rW r\ r] r^ re r r rm rl � staticmethodrq r3 r4 ru r} r� rU rv r{ r� rL rR r@ r r )r% r rA s sD
%
#
8''
rA )r r
r r )r? � collectionsr r � itertoolsr Zzmqr r r r0 r ZPollerr
r1 rA r r r r �<module> s
\