File: //opt/saltstack/salt/lib/python3.10/site-packages/salt/transport/__pycache__/zeromq.cpython-310.pyc
o
�N�g3� � @ s� d Z ddlZddlZddlZddlZddlZddlZddlZddlm Z ddl
Z ddl Z ddl
Z ddlZddlZddlZddlZddlZddlZddlZddlZddlZddlZddlZddlmZ ddl m Z m Z ddlm Z m Z m Z zddl!Z dZ"W n e#y� dZ"Y nw e�$e%�Z&dd d
�Z'G d d � d ej(j)j*�Z*G d
d� dej(j)j+�Z,dd� Z-G dd� d�Z.G dd� d�Z/G dd� dej(j)j0�Z1G dd� dej(j)j2�Z2dS )z
Zeromq transport classes
� N)�randint)� ipaddress)�
SaltException�SaltReqTimeoutError)�LIBZMQ_VERSION_INFO�ZMQ_VERSION_INFO�zmqTFc C s� ddl m} dj|| �|d�}|s|rntdkr_tdkr_|r.|r.dj||�||| �|d�}|S |r@|s@d j||�|| �|d
�}|S |r]|s]t�| �jd krNd n|d
�}dj|||| �|d�}|S t� d� t� d� t� d� |S )aR
Return the ZeroMQ URI to connect the Minion to the Master.
It supports different source IP / port, given the ZeroMQ syntax:
// Connecting using a IP address and bind to an IP address
rc = zmq_connect(socket, "tcp://192.168.1.17:5555;192.168.1.1:5555"); assert (rc == 0);
Source: http://api.zeromq.org/4-1:zmq-tcp
r )�
ip_bracketz tcp://{master_ip}:{master_port})� master_ip� master_port)� � � )� r r
z9tcp://{source_ip}:{source_port};{master_ip}:{master_port})� source_ip� source_portr
r z-tcp://{source_ip}:0;{master_ip}:{master_port})r r
r r z0.0.0.0z::z6tcp://{ip_any}:{source_port};{master_ip}:{master_port})�ip_anyr r
r zAUnable to connect to the Master using a specific source IP / portz9Consider upgrading to pyzmq >= 16.0.1 and libzmq >= 4.1.6zVSpecific source IP / port for connecting to master returner port: configuraion ignored)
Zsalt.utils.networkr �formatr r r �
ip_address�version�log�warning)r
r r r r �
master_urir � r �I/opt/saltstack/salt/lib/python3.10/site-packages/salt/transport/zeromq.py�_get_master_uri, sV ���&�� �������
�r c s� e Zd ZdZdZ� fdd�Zdd� Zdd� Zd d
� Ze j
j j j
dd d
��Zedd� �Ze j
j j j
dd� �Zedd� �Zdd� Ze j
j j j
dd� �Z� ZS )�
PublishClientzw
A transport channel backed by ZeroMQ for a Salt Publisher to use to
publish commands to connected minions
�zeromqc sv t � j||fi |�� || _|| _t�tjj� | jd ���
� | _ d| _ t
�� | _| j�t
j�| _| jd r^| j�t
jd� | j�d�dkrO| j�t
jd� n| j�t
jtjj� | j �� n| j�t
jd� | j�t
jtjj� | jd �� tt
d �r�| j�t
j| jd
� | j�t
j| jd � | j�t
j| jd � | j�t
j| jd
� | jd }| jd r�t | jd | jd | jd �}t � d| jd | jd | jd |� t � d|� | j�t
j |� tt
d��rt � d| jd | jd � | j�t
j | jd � | jd du �sd| jd v �r tt
d��r | j�t
j!d� t"�r7| jd �r9t#| j�| _$| j$�%| j� d S d S d S ) N�idF�
zmq_filtering� broadcast�__role�syndic� syndic� �
TCP_KEEPALIVE�
tcp_keepalive�tcp_keepalive_idle�tcp_keepalive_cnt�tcp_keepalive_intvlZ
recon_defaultZrecon_randomizeZ recon_maxz?Generated random reconnect delay between '%sms' and '%sms' (%s)z#Setting zmq_reconnect_ivl to '%sms'�RECONNECT_IVL_MAXz'Setting zmq_reconnect_ivl_max to '%sms'�ipv6T�:r
�IPV4ONLYr � zmq_monitor)&�super�__init__�opts�io_loop�hashlib�sha1�salt�utils� stringutils�to_bytes� hexdigest�hexid�_closingr �Context�context�socketZSUB�_socket�
setsockoptZ SUBSCRIBE�getZIDENTITY�hasattrr% �TCP_KEEPALIVE_IDLE�TCP_KEEPALIVE_CNT�TCP_KEEPALIVE_INTVLr r �debugZ
RECONNECT_IVLr* r- �HAS_ZMQ_MONITOR�ZeroMQSocketMonitor�_monitor�
start_io_loop)�selfr1 r2 �kwargsZ recon_delay�� __class__r r r0 p sz ��
��
� � �
�� �"� �zPublishClient.__init__c C s� | j du rd S d| _ t| d�r | jd ur | j�� d | _t| d�r(| j�d� n t| d�r3| j�d� t| d�rE| jjdu rG| j� � d S d S d S )NTrI �_streamr r? r= F)
r; rB rI �stoprO �closer? r= �closed�term�rK r r r rQ � s
�zPublishClient.closec C � | S �Nr rT r r r � __enter__� � zPublishClient.__enter__c C � | � � d S rV �rQ �rK �exc_typeZexc_valZexc_tbr r r �__exit__� � zPublishClient.__exit__Nc C sP d| _ || _t�d| j� t�d| | j� | j�| j� |d ur&|d� d S d S )NTzCConnecting the Minion to the Master publish port, using the URI: %sz%r connecting to %s)�_connect_called� publish_portr rF �
master_pubr? �connect)rK r` Zconnect_callbackZdisconnect_callbackr r r rb � s � �zPublishClient.connectc C s( t | jd | j| j�d�| j�d�d�S )z0
Return the master publish port
r
r Zsource_publish_port�r r )r r1 r` rA rT r r r ra � s
�zPublishClient.master_pubc C s� t |�}|dkrtj�|d �}nH|dkrPtjj�|d �}| j�d�dkr-|d| j fvs9| j�d�dkrG|dvrGt
� d|� tj j
j�d ��tj�|d �}n td
�t |����tj j
j�|��) z�
Take the zmq messages, decrypt/decode them into a payload
:param list messages: A list of messages to be decoded
r
r � r! r" � broadcast)re r" z(Publish received for not this minion: %sNz@Invalid number of messages ({}) in zeromq pubmessage from master)�lenr5 �payload�loadsr6 r7 Zto_strr1 rA r: r rF �ext�tornado�gen�Return� Exceptionr )rK �messagesZ messages_lenrg Zmessage_targetr r r �_decode_messages� s$ ��z PublishClient._decode_messagesc C s( t | d�stjjj| j| jd�| _| jS )zI
Return the current zmqstream, creating one if necessary
rO �r2 )rB r � eventloop� zmqstream� ZMQStreamr? r2 rO rT r r r �stream s
�zPublishClient.streamc
C sL z| j �|�W S ty% } z|du r t|�dkr W Y d}~dS � d}~ww )z�
Register a callback for received messages (that we didn't initiate)
:param func callback: A function which should be called when data is received
NzStream is closed)rt �on_recv�OSError�str)rK �callback�excr r r ru s ��zPublishClient.on_recvc C s | j j|dd� d S )NT)Znoblock)rt �send�rK �msgr r r rz $ s zPublishClient.send�NN)�__name__�
__module__� __qualname__�__doc__�ttyper0 rQ rW r] r5 ri rj rk � coroutinerb �propertyra ro rt ru rz �
__classcell__r r rM r r h s$ K
!
r c @ sp e Zd Zdd� Zdd� Zdd� Zdd� Zd d
� Zd d � Ze j
j j j
d
d� �Zdd� Zdd� Zdd� Zdd� ZdS )�
RequestServerc C s || _ d| _d | _d | _d S )NF)r1 r; rI �
_w_monitor�rK r1 r r r r0 * s
zRequestServer.__init__c
C s( | � � t�| jd �}djdi | j��| _|�tj�| _| j� tj
d� | jd du r9t td�r9| j� tj d� | j� tj
| j�dd �� | �� |�tj�| _| j� tj
d� | jd
rutjj�� sut�d | jd
� t�| jd
� | j�d d
�dkr�d�| j�dd��| _nd�tj� | jd d��| _t�d� t�d| j� | j� | j� t�d| j� | j� | j� | j�d d
�dkr�t� tj� | jd d�d� | jj s�| jj r�n4z t� tj!| j| j� W n& tj"�y } z
|j#t#j$kr�W Y d}~qЂ d}~w t%t&f�y Y nw q�|�'� dS )zA
Multiprocessing target for the zmq queue device
Zworker_threadsz tcp://{interface}:{ret_port}���r+ Tr- r � zmq_backlog�� Zmworker_queue_nicenessz$setting mworker_queue niceness to %d�ipc_mode� �tcp�tcp://127.0.0.1:{}�tcp_master_workers� �ipc://{}�sock_dir� workers.ipcz*Setting up the master communication serverzReqServer clients %szReqServer workers %s� Nr )(� _RequestServer__setup_signalsr r< r1 r Zurir> ZROUTER�clientsr@ �LINGERrB r- �BACKLOGrA �_start_zmq_monitorZDEALER�workersr5 r6 �platformZ
is_windowsr �info�os�nice�w_uri�path�join�bind�chmodrR ZdeviceZQUEUE�ZMQError�errnoZEINTR�KeyboardInterrupt�
SystemExitrS )rK r= ry r r r �
zmq_device0 s\ � ��
��� zRequestServer.zmq_devicec C s | j rdS t�dt�� � d| _ t| dd�dur | j�� d| _t| dd�dur0| j�� d| _t | d�r@| j
j du r@| j
� � t | d�rP| j
j du rP| j
� � t | d �rZ| j� � t | d
�rj| jj du rj| j� � t | d �r|| jj du r~| j�� dS dS dS ) z4
Cleanly shutdown the router socket
Nz$MWorkerQueue under PID %s is closingTrI r� r� Fr� rt r? r= )r; r r� r� �getpid�getattrrI rP r� rB r� rR rQ r� rt r? r= rS rT r r r rQ g s*
�zRequestServer.closec C s |j | jdd� dS )z�
Pre-fork we need to create the zmq router device
:param func process_manager: An instance of salt.utils.process.ProcessManager
Z MWorkerQueue)�nameN)� add_processr� �rK Zprocess_managerr r r �pre_fork� s zRequestServer.pre_forkc C sN t r#| jd r%t�d� t| j�| _tj| jj d��
� t�d� dS dS dS )zM
Starts ZMQ monitor for debugging purposes.
:return:
r. zStarting ZMQ monitor)�targetz$ZMQ monitor has been started startedN) rG r1 r rF rH r? r� � threading�Thread�
start_poll�startrT r r r r� � s
�z RequestServer._start_zmq_monitorc C s
t �d�}|�t j�| _| j�t jd� | �� | j� dd�dkr-d�
| j� dd��| _ nd �
t j
�| jd
d ��| _ t�d | j � | j�| j � | j� dd�dkrnt j
�t j
�| jd
d ��rnt �t j
�| jd
d �d
� t jjj| j|d�| _|| _| j�| j� dS )ad
After forking we need to create all of the local sockets to listen to the
router
:param func message_handler: A function to called to handle incoming payloads as
they are picked up off the wire
:param IOLoop io_loop: An instance of a Tornado IOLoop, to handle event scheduling
r
r� r� r� r� r� r� r� r� r� r� zWorker binding to socket %sr� rp N)r r< r> ZREPr? r@ r� r� r1 rA r r� r� r� r� r r� rb �isfiler� rq rr rs rt �message_handlerZon_recv_stream�handle_message)rK r� r2 r= r r r � post_fork� s(
��� zRequestServer.post_forkc c sb � z| � |�}W n tjjy | j�| �ddi�� Y d S w | �|�V }| j�| �|�� d S )Nr| zbad load)�decode_payloadr5 �
exceptionsZSaltDeserializationErrorrt rz �encode_payloadr� )rK rt rg Zreplyr r r r� � s �� z RequestServer.handle_messagec C s t j�|�S rV )r5 rg �dumps�rK rg r r r r� � r^ z RequestServer.encode_payloadc C s$ t � t j| j� t � t j| j� d S rV )�signal�SIGINT�_handle_signals�SIGTERMrT r r r Z__setup_signals� s z RequestServer.__setup_signalsc C sb | j j� d�}|tjkr|d7 }n |tjkr|d7 }|d7 }t�|� | �� t� t
j j j
� d S )Nz received a r� r� z . Exiting)rN r~ r� r� r� r rF rQ �sys�exitr5 �defaultsZ exitcodes�EX_OK)rK �signumZsigframer| r r r r� � s
z RequestServer._handle_signalsc C s t j�|d �}|S )Nr )r5 rg rh r� r r r r� � s z RequestServer.decode_payloadN)r~ r r� r0 r� rQ r� r� r� r5 ri rj rk r� r� r� r� r� r� r r r r r� ) s 7
!
r� c C s� t td�r=|r?d|v r| �tj|d � d|v r!| �tj|d � d|v r.| �tj|d � d|v rA| �tj|d � dS dS dS dS )a�
Ensure that TCP keepalives are set as specified in "opts".
Warning: Failure to set TCP keepalives on the salt-master can result in
not detecting the loss of a minion when the connection is lost or when
its host has been terminated without first closing the socket.
Salt's Presence System depends on this connection status to know if a minion
is "present".
Warning: Failure to set TCP keepalives on minions can result in frequent or
unexpected disconnects!
r% r& r' r( r) N)rB r r@ r% rC rD rE )Z
zmq_socketr1 r r r �_set_tcp_keepalive� s
�r� c @ sd e Zd ZdZddd�Zdd� Zdd � Zd
d � Zej j
j j dd d
��Z
dd� Zej j
j j dd� �ZdS )�AsyncReqMessageClienta�
This class wraps the underlying zeromq REQ socket and gives a future-based
interface to sending and recieving messages. This works around the primary
limitation of serialized send/recv on the underlying socket by queueing the
message sends in this class. In the future if we decide to attempt to multiplex
we can manage a pool of REQ/REP sockets-- but for now we'll just do them in serial
r Nc C sr || _ || _|| _|du rtjjjj�� | _ n|| _ t
j j �
� | _g | _d| _i | _tjjj�� | _t�� | _dS )a�
Create an asynchronous message client
:param dict opts: The salt opts dictionary
:param str addr: The interface IP address to bind to
:param int linger: The number of seconds to linger on a ZMQ socket. See
http://api.zeromq.org/2-1:zmq-setsockopt [ZMQ_LINGER]
:param IOLoop io_loop: A Tornado IOLoop event scheduler [tornado.ioloop.IOLoop]
NF)r1 �addr�lingerr5 ri rj �ioloop�IOLoop�currentr2 r rq �futurer< r= Z
send_queuer; Z_send_future_map�locks�Lock�lockr� � get_ident�ident)rK r1 r� r� r2 r r r r0 � s
z AsyncReqMessageClient.__init__c C s t | d�r
| jr
d S | �� d S )Nr> )rB r> � _init_socketrT r r r rb s z AsyncReqMessageClient.connectc C sT | j rd S d| _ t| d�r| jd ur| j�d� d | _| jjdu r(| j�� d S d S )NTr> r F)r; rB r> rQ r= rR rS rT r r r rQ s �zAsyncReqMessageClient.closec C s� | j �tj�| _ttd�r| j�tjd� t| j| j� | j �
d�r=ttd�r0| j�tj d� n
ttd�r=| j�tj d� | j�tj
| j� | j�| j � d S )Nr* i� ztcp://[�IPV6r
r- r )r= r> r ZREQrB r@ r* r� r1 r� �
startswithr� r- r� r� rb rT r r r r� ' s
z"AsyncReqMessageClient._init_socketc # s� � t jjj�� }t j�|�}� dur � �fdd�}|�|� �j� d�du r(d}|dur5�j
� |�j |�}�j
�
�j||� |V }t jjj�|��)zY
Return a future which will be completed when the message has a response
Nc s | � � }�j�� |� d S rV )�resultr2 Z add_callback)r� �response�rx rK r r �
handle_futureC s z1AsyncReqMessageClient.send.<locals>.handle_futureZ detect_modeTr
)r5 ri rj Z
concurrentZFuturerg r� Zadd_done_callbackr1 rA r2 Z
call_later�_timeout_messageZspawn_callback�
_send_recvrk rl )rK �message�timeoutrx r� r� Z send_timeout�recvr r� r rz 8 s �
�zAsyncReqMessageClient.sendc C s |� � s
|�td�� d S d S )NzMessage timed out)�done�
set_exceptionr )rK r� r r r r� W s �z&AsyncReqMessageClient._timeout_messagec c s � za| j �� V �> | j�|�V z| j�� V }W n& tjjjy> } z|� � s,|�
|� W Y d }~W d � W d S d }~ww W d � n1 sIw Y |� � s`t j �
|�}|�|� W d S W d S ty� } z|� � sz|�
|� W Y d }~d S W Y d }~d S d }~ww rV )r� �acquirer> rz r� r rq r� ZCancelledErrorr� r� r5 rg rh Z
set_resultrm )rK r� r� r� ry �datar r r r� [ s2 �
��� � ����z AsyncReqMessageClient._send_recv)r Nr} )r~ r r� r� r0 rb rQ r� r5 ri rj rk r� rz r� r� r r r r r� � s
r� c @ sD e Zd ZdZdd� Zdd� Zdd� Zedd � �Zd
d � Z d d
� Z
dS )rH Nc C s || _ | j �� | _d| _dS )z�
Create ZMQ monitor sockets
More information:
http://api.zeromq.org/4-0:zmq-socket-monitor
N)r? Zget_monitor_socket�_monitor_socket�_monitor_stream)rK r> r r r r0 r s
z ZeroMQSocketMonitor.__init__c C s2 t �d� tjjj| j|d�| _| j�| j � d S )N�Event monitor start!rp )
r �tracer rq rr rs r� r� ru �monitor_callback)rK r2 r r r rJ } s
�z!ZeroMQSocketMonitor.start_io_loopc C s| t �d� z*| jd ur'| j�� r-| j�� }| �|� | jd ur*| j�� sW d S W d S W d S W d S ttjj fy= Y d S w )Nr� )
r r� r� �pollZrecv_multipartr� �AttributeErrorr �errorZContextTerminatedr{ r r r r� � s
,��z ZeroMQSocketMonitor.start_pollc C sD t jd u r i }tt�D ]}|�d�rtt|�}|||< q |t _t jS )NZEVENT_)rH � _ZeroMQSocketMonitor__EVENT_MAP�dirr r� r� )rK � event_mapr� �valuer r r r� � s
�z ZeroMQSocketMonitor.event_mapc C sJ t jj�|�}| j|d |d< t�d|� |d t jkr#| �� d S d S )N�event� descriptionzZeroMQ event: %s) r r6 �monitorZparse_monitor_messager� r rF ZEVENT_MONITOR_STOPPEDrP )rK r| Zevtr r r r� � s �z$ZeroMQSocketMonitor.monitor_callbackc C sf | j d u rd S z| j �� W n
tjy Y nw d | _ d | _| jd ur,| j�� d | _t�d� d S )NzEvent monitor done!) r? Zdisable_monitorr �Errorr� r� rQ r r� rT r r r rP � s
�
zZeroMQSocketMonitor.stop) r~ r r� r� r0 rJ r� r� r� r� rP r r r r rH o s
rH c @ s� e Zd ZdZe�� Zdd� Zdd� Z d!dd�Z e
d d
� �Z e
d d � �Z e
jjjjd"d
d��Zdd� Ze
dd� �Zdd� Zdd� Zdd� Ze
dd� �Zdd � Zd d � Zd d � ZdS )#�
PublishServerzD
Encapsulate synchronous operations for a publisher channel
c C s
|| _ d S rV �r1 r� r r r r0 � s
zPublishServer.__init__c C s t jjj�d�S )N� )r5 ri rj rk �sleeprT r r r rb � s zPublishServer.connectNc
s4 t jjj�� }|�� || _t�d�}|� tj
�}t |�}|� |� t
|| j� tjj�|� | _}z|�tj| j�dd�� W n% ttjjfye |�tj| j�dd�� |�tj| j�dd�� Y nw | jd du ryttd�ry|�tj d� |�tj | j�dd�� |�tj d � |� tj �}tjj�|�}|�tj d � t j j!�"| j#� t$�%d
| j&� |�'| j&� t$�%d | j#� t j j(�)d �� |�'| j#� W d
� n1 s�w Y t jjj*j+� fdd��} |�,| � z&z|�-� W n t.y� Y n w W |�/� |�/� d
S W |�/� |�/� d
S |�/� |�/� w )z�
This method represents the Publish Daemon process. It is intended to be
run in a thread or process as it creates and runs its own ioloop.
r
Zpub_hwmr� r+ Tr- r r� r� z!Starting the Salt Publisher on %sz Starting the Salt Puller on %s� Nc
3 sb � z| D ]
}t j�|�}� |�V qW d S ty0 } ztjd|tjd� W Y d }~d S d }~ww )Nz Un-handled error in publisher %s)Zexc_info_on_loglevel)r5 rg rh rm r r� �logging�DEBUG)Zpackages�packagerg ry ��publish_payloadr r ru � s � ����z-PublishServer.publish_daemon.<locals>.on_recv)0r5 ri rj r� r� Z make_currentr2 r r< r> ZPUBrH rJ r� r1 rq rr rs � dpub_sockr@ ZHWMrA r� r� r� ZSNDHWMZRCVHWMrB r- r� r� ZPULLr6 r Zcheck_ipc_path_max_len�pull_urir r� �pub_urir� �filesZ set_umaskrk r� ru r� r� rQ )
rK r� Zpresence_callbackZremove_presence_callbackr� r= �pub_sockr� Z pull_sockru r r� r �publish_daemon� sZ
� �
�� � �
z PublishServer.publish_daemonc C sH | j �dd�dkrd�| j �dd��}|S d�tj�| j d d ��}|S )
Nr� r� r� r� �tcp_master_publish_pull� r� r� �publish_pull.ipc)r1 rA r r� r� r� )rK r� r r r r� s ���zPublishServer.pull_uric C s dj di | j��S )Nz tcp://{interface}:{publish_port}r )r r1 rT r r r r
s zPublishServer.pub_uric c s � t j�|�}| jd rm|rV|D ](}t�d| j� t jj� t
� t jj� |��� � �}| j
�||g�V t�d� q| j�d�rTt�d� | j
�d|g�V t�d� d S d S t�d| j� | j
�d |g�V t�d
� d S t�d | j� | j
�|�V t�d � d S )
Nr z'Sending filtered data over publisher %szFiltered data has been sentZ
order_mastersz Sending filtered data to syndicr# z%Filtered data has been sent to syndicz*Sending broadcasted data over publisher %sr z Broadcasted data has been sentz-Sending ZMQ-unfiltered data over publisher %sz Unfiltered data has been sent)r5 rg r� r1 r r� r r6 r7 r8 r3 r4 r9 r� Zsend_multipartrA rz )rK rg Z
topic_listZtopicZhtopicr r r r� s, �
�
�z PublishServer.publish_payloadc C s |j | j| jfd� dS )a.
Do anything necessary pre-fork. Since this is on the master side this will
primarily be used to create IPC channels and create our daemon process to
do the actual publishing
:param func process_manager: A ProcessManager, from salt.utils.process.ProcessManager
)�argsN)r� r r� r� r r r r� / s
�zPublishServer.pre_forkc C s z| j jW S ty Y dS w )z�
This thread's zmq publisher socket. This socket is stored on the class
so that multiple instantiations in the same thread will re-use a single
zmq socket.
N)�
_sock_data�sockr� rT r r r r < s
�zPublishServer.pub_sockc C s� | j r| �� t�� }|�tj�| j_| j �tj d� | j
� dd�dkr/d� | j
� dd��}n
d� t
j�| j
d d
��}t�d |� | j �|� | jjS ) z�
Create and connect this thread's zmq socket. If a publisher socket
already exists "pub_close" is called before creating and connecting a
new socket.
r� r� r� r� r� r r r� r� r z Connecting to pub server: %s)r � pub_closer r< r> ZPUSHr r r@ r� r1 rA r r� r� r� r rF rb )rK �ctxr� r r r � pub_connectH s �� zPublishServer.pub_connectc C s, t | jd�r| jj�� t| jd� dS dS )zn
Disconnect an existing publisher socket and remove it from the local
thread's cache.
r N)rB r r rQ �delattrrT r r r r
_ s �zPublishServer.pub_closec K s4 | j s| �� tj�|�}| j �|� t�d� dS )z�
Publish "load" to minions. This send the load to the publisher daemon
process with does the actual sending to minions.
:param dict load: A load to be sent across the wire to minions
z Sent payload to publish daemon.N)r r r5 rg r� rz r rF )rK rg rL Z
serializedr r r �publishh s
zPublishServer.publishc C s | j �dd�S )Nr F)r1 rA rT r r r �
topic_supportu s zPublishServer.topic_supportc C rY rV )r
rT r r r rQ y r^ zPublishServer.closec C rU rV r rT r r r rW | rX zPublishServer.__enter__c C rY rV rZ r[ r r r r] r^ zPublishServer.__exit__r} rV ) r~ r r� r� r� �localr r0 rb r r� r� r r5 ri rj rk r� r� r� r r r
r r rQ rW r] r r r r r� � s2
�D
r� c s^ e Zd ZdZ� fdd�Zejjjj dd� �Z
ejjjj d
dd��Z d d
� Z e
d d � �Z� ZS )�
RequestClientr c s@ t � �||� || _| �|�}t| j||d�| _d| _d| _d S )Nrp F)r/ r0 r1 �get_master_urir� �message_clientr; r_ )rK r1 r2 r rM r r r0 � s
�
zRequestClient.__init__c C s d| _ | j�� d S �NT)r_ r rb rT r r r rb � s zRequestClient.connect�< c c s. � | � � V | jj||d�V }tjjj�|��)N)r� )rb r rz r5 ri rj rk rl )rK �loadr� �retr r r rz � s �
zRequestClient.sendc C s | j rd S d| _ | j�� d S r )r; r rQ rT r r r rQ � s zRequestClient.closec C sD d| v r| d S d| v r t | d | d | �d�| �d�d�S td��)Nr r
r r Zsource_ret_portrc z5ReqChannel: missing master_uri/master_ip in self.opts)r rA r r� r r r r � s �z RequestClient.get_master_uri)r )r~ r r� r� r0 r5 ri rj rk r� rb rz rQ � staticmethodr r� r r rM r r � s
r r} )3r� r� r3 r� r� r� r� r� �randomr Z zmq.errorr Zzmq.eventloop.futureZzmq.eventloop.zmqstreamZsalt.ext.tornador5 Zsalt.ext.tornado.concurrentZsalt.ext.tornado.genZsalt.ext.tornado.ioloopZsalt.ext.tornado.locksZ salt.payloadZsalt.transport.baseZsalt.utils.filesZsalt.utils.processZsalt.utils.stringutilsZsalt.utils.zeromqZ salt._compatr Zsalt.exceptionsr r r r Zzmq.utils.monitorrG � ImportError� getLoggerr~ r r Z transport�baser ZDaemonizedRequestServerr� r� r� rH ZDaemonizedPublishServerr� r r r r r �<module> sX �
< B 1}A T