File: //opt/saltstack/salt/lib/python3.10/site-packages/salt/transport/__pycache__/tcp.cpython-310.pyc
o
�N�gT� � @ s6 d Z ddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl
Z ddl Z ddl
Z ddlZ ddlZ ddlZ ddlZ ddlZ ddlZ ddlZ ddlZ ddlZ ddlZ ddlZ ddlZ ddlZ ddlm Z m Z ddl m Z ddl m!Z! e j"j#�$� r�dZ%ndZ%e%r�ddl&Z e�'e(�Z)G dd � d e*�Z+d
d � Z,d d
� Z-dd� Z.G dd� de!�Z/G dd� d�Z0G dd� de j1j2j3�Z4G dd� de j1j2j5�Z6G dd� de j7j8j9j:�Z;G dd� de;�Z<G d d � d e j7j8j=j>�Z?G d d � d �Z@G d d!� d!�ZAG d"d#� d#e j7j8j9j:�ZBG d$d%� d%e j1j2jC�ZDG d&d'� d'e j1j2jE�ZFdS )(zh
TCP transport classes
Wire protocol: "len(payload) msgpack({'head': SOMEHEADER, 'body': SOMEBODY})"
� N)�SaltClientError�SaltReqTimeoutError)�
ip_bracket)�SignalHandlingProcessTFc @ s e Zd ZdZdS )� ClosingError� N)�__name__�
__module__� __qualname__�__doc__� r r �F/opt/saltstack/salt/lib/python3.10/site-packages/salt/transport/tcp.pyr 2 s r c C s& t j}| �dd�r t j}t � |t j�S )NZipv6F)�socket�AF_INET�get�AF_INET6� SOCK_STREAM)�opts�familyr r r
� _get_socket6 s r c C s t | d dd�t| | �fS )NZ interfaceT��strip)r �int)r Z port_typer r r
�_get_bind_addr= s
�r c C sb t td�r�|�dd�r�| �tjtjd� t td�rgt td�r3|�dd�}|d kr3| �tjtjt|�� t td
�rM|�d d�}|d krM| �tjtj t|�� t td �rg|�d
d�}|d krg| �tjtj
t|�� t td�r�|�dd�}|�d
d�}|d ks�|d kr�|d kr�d}|d kr�d}| � tj dt|d �t|d �f� dS dS dS | �tjtjd � dS dS )z<
Ensure that TCP keepalives are set for the socket.
� SO_KEEPALIVEZ
tcp_keepaliveF� �SOL_TCP� TCP_KEEPIDLE�tcp_keepalive_idle���r � TCP_KEEPCNT�tcp_keepalive_cnt�
TCP_KEEPINTVL�tcp_keepalive_intvl�SIO_KEEPALIVE_VALSi i� N)
�hasattrr r �
setsockopt�
SOL_SOCKETr r r r r r" �ioctlr$ )�sockr r r! r# r r r
�_set_tcp_keepaliveD sV
�
�
�
����r* c s< e Zd ZdZdZ� fdd�Zdd� Zdd� Zd d
� Z� Z S ) �LoadBalancerServera
Raw TCP server which runs in its own process and will listen
for incoming connections. Each incoming connection will be
sent via multiprocessing queue to the workers.
Since the queue is shared amongst workers, only one worker will
handle a given connection.
� c s( t � jdi |�� || _|| _d | _d S )Nr )�super�__init__r � socket_queue�_socket)�selfr r/ �kwargs�� __class__r r
r. � s
zLoadBalancerServer.__init__c C s0 | j d ur| j �tj� | j �� d | _ d S d S �N)r0 �shutdownr � SHUT_RDWR�close�r1 r r r
r8 � s
�zLoadBalancerServer.closec C � | � � d S r5 �r8 r9 r r r
�__del__� � zLoadBalancerServer.__del__c
C s� t | j�| _| j�tjtjd� t| j| j� | j�d� | j� t
| jd�� | j� | j � z| j�
� \}}| j�||fdd� W n ty` } ztjjj�|�tjkr[W Y d}~q.� d}~ww q/)z)
Start the load balancer
r �ret_portTN)r r r0 r& r r' � SO_REUSEADDRr* � setblocking�bindr �listen�backlog�acceptr/ �put�OSError�salt�ext�tornado�utilZerrno_from_exception�errnoZ ECONNABORTED)r1 �
connection�address�er r r
�run� s" ���zLoadBalancerServer.run)
r r r
r rC r. r8 r< rO �
__classcell__r r r3 r
r+ � s
r+ c @ s& e Zd ZdZeddd��Zdd� ZdS ) �ResolverF�
c C s t jjjjjd|d� d| _d S )Nz)salt.ext.tornado.netutil.ThreadedResolver)� num_threadsT)rG rH rI ZnetutilrQ Z configure�_resolver_configured)�clsrS r r r
�_config_resolver� s �
zResolver._config_resolverc O s | j s | �� d S d S r5 )rT rV )r1 �argsr2 r r r
r. � s �zResolver.__init__N)rR )r r r
rT � classmethodrV r. r r r r
rQ � s
rQ c s� e Zd ZdZdZ� fdd�Zdd� Zejj j
j ddd ��Z ejj j
j d
d � �Z
ejj j
j d d
� �Zdd� Zdd� Zdd� Z� ZS )� TCPPubClientz&
Tornado based TCP Pub Client
�tcpc s@ t � j||fi |�� || _|| _d | _d| _d| _t� | _d S �NF) r- r. r �io_loop�message_client� connected�_closingrQ �resolver)r1 r r\ r2 r3 r r
r. � s zTCPPubClient.__init__c C s2 | j rd S d| _ | jd ur| j�� d | _d S d S �NT�r_ r] r8 r9 r r r
r8 � s
�zTCPPubClient.closeNc c s^ � d| _ || _t| j| jd t| j�| j||| j�d�| j�d�d�| _| j�� V d| _ d S )NTZ master_ip� source_ipZsource_publish_port)r\ �connect_callback�disconnect_callbackrc � source_port)
�_connect_called� publish_port�
MessageClientr r r\ r r] �connectr^ )r1 rh rd re r r r
rj � s �
�
zTCPPubClient.connectc C s< t |t�stjj�|�}tjj�|�}n|}tj j
j � |��r5 )
�
isinstance�dictrG �utils�msgpack�loads� transport�frame�decode_embedded_strsrH rI �gen�Return)r1 �messages�bodyr r r
�_decode_messages� s
z TCPPubClient._decode_messagesc c s � | j j�|�V d S r5 )r] �_stream�write)r1 �msgr r r
�send s �zTCPPubClient.sendc C s | j �|�S )z.
Register an on_recv callback
)r] �on_recv)r1 �callbackr r r
r| s zTCPPubClient.on_recvc C � | S r5 r r9 r r r
� __enter__
� zTCPPubClient.__enter__c C r: r5 r; )r1 �exc_typeZexc_valZexc_tbr r r
�__exit__ r= zTCPPubClient.__exit__�NN)r r r
r �ttyper. r8 rG rH rI rs � coroutinerj rw r{ r| r r� rP r r r3 r
rY � s
rY c @ sn e Zd ZdZdZdd� Zedd� �Zdd� Zd d
� Z d d � Z
d
d� Z dd� Z e
jjjjddd��Zdd� ZdS )� TCPReqServerzc
Tornado based TCP Request/Reply Server
:param dict opts: Salt master config options.
� c C s || _ d | _d | _d S r5 )r r0 �
req_server�r1 r r r r
r. s
zTCPReqServer.__init__c C s | j S r5 )r0 r9 r r r
r # s zTCPReqServer.socketc
C s� | j d ur6z | j �tj� W n ty( } z|jtjkr n� W Y d }~nd }~ww | jd u r3| j �� d | _ | jd urhz| j�� W n tyb } z|jdkrP� t �
dt |�� W Y d }~nd }~ww d | _d S d S )N� z4TCPReqServerChannel close generated an exception: %s) r0 r6 r r7 rF rK �ENOTCONNr� r8 �log� exception�str)r1 �excr r r
r8 ' s4
���
���
�zTCPReqServer.closec C r~ r5 r r9 r r r
r E r� zTCPReqServer.__enter__c G r: r5 r; )r1 rW r r r
r� H r= zTCPReqServer.__exit__c C s� t rt�� | _|jt| j| jfdd� dS tjj �
� sDt | j�| _ | j �
tjtjd� t| j | j� | j �d� | j �t| jd�� dS dS )zB
Pre-fork we need to create the zmq router device
r+ )rW �namer r r> N)�USE_LOAD_BALANCER�multiprocessing�Queuer/ � add_processr+ r rG rm �platform�
is_windowsr r0 r& r r' r? r* r@ rA r �r1 Zprocess_managerr r r
�pre_forkK s
� �zTCPReqServer.pre_forkc C s� || _ tjj�|��j tr t| j| j| j �
d�d�| _ nPtjj �
� rIt| j �| _| j�tjtjd� t| j| j � | j�d� | j�t| j d�� t| j| j �
d�|d�| _ | j �| j� | j�| j� W d � dS W d � dS 1 sww Y dS )z�
After forking we need to create all of the local sockets to listen to the
router
message_handler: function to call with your payloads
�ssl�� ssl_optionsr r r> )r� r\ N) �message_handlerrG rm � asynchronous�current_ioloopr� �LoadBalancerWorkerr/ �handle_messager r r� r� r� r r0 r& r r' r? r* r@ rA r �SaltMessageServer�
add_socketrB rC )r1 r� r\ r r r
� post_fork] s0
�
��"�zTCPReqServer.post_forkNc c s4 � | � |�}| �|�V }|�tjjj||d�� d S )N��header)�decode_payloadr� ry rG rp rq � frame_msg)r1 �stream�payloadr� Zreplyr r r
r� | s �
zTCPReqServer.handle_messagec C � |S r5 r )r1 r� r r r
r� � r� zTCPReqServer.decode_payloadr5 )r r r
r rC r. �propertyr r8 r r� r� r� rG rH rI rs r� r� r� r r r r
r� s
r� c sP e Zd ZdZ� fdd�Zejjjj ejjj
j fdd��Z dd� Z
dd � Z� ZS )
r� z{
Raw TCP server which will receive all of the TCP streams and re-assemble
messages that are sent through to us
c sJ |� dd �p
tjjjj�� }d| _t� j |i |�� || _
g | _ || _ d S )Nr\ F)
�poprG rH rI �ioloop�IOLoop�currentr_ r- r. r\ �clientsr� )r1 r� rW r2 r\ r3 r r
r. � s �
zSaltMessageServer.__init__c
c s� � t �d|� | j�||f� tjj�� }z+ |jddd�V }|� |� |D ]}tj
j � |�}|d }| j
�| j||d |� q&q |yW t �d|� | �||f� Y d
S ty| } zt jd|dd � | �||f� |�� W Y d
}~d
S d
}~ww ) zP
Handle incoming streams and add messages to the incoming queue
zReq client %s connectedT� ��partial�headrv zreq client disconnected %sz other master-side exception: %s��exc_infoN)r� �tracer� �appendrG rm rn �Unpacker�
read_bytes�feedrp rq rr r\ �spawn_callbackr� �
remove_client� Exceptionr8 ) r1 r� rM Z_StreamClosedError�unpacker�
wire_bytes�
framed_msgr� rN r r r
�
handle_stream� s0 �
�� ��z SaltMessageServer.handle_streamc C s2 z | j �|� W d S ty t�d� Y d S w )Nz/Message server client was not in list to remove)r� �remove�
ValueErrorr� r� �r1 �clientr r r
r� � s
�z SaltMessageServer.remove_clientc
C sx | j rdS d| _ | jD ]}|\}}|�� | �|� q z| �� W dS ty; } z
|jdkr0� W Y d}~dS d}~ww )z"
Close the server
NTr� )r_ r� r8 r� �stoprF rK )r1 �itemr� rM r� r r r
r8 � s
���zSaltMessageServer.close)r r r
r r. rG rH rI rs r� �iostream�StreamClosedErrorr� r� r8 rP r r r3 r
r� � s
� r� c s4 e Zd ZdZ� fdd�Z� fdd�Zdd� Z� ZS )r� z�
This will receive TCP connections from 'LoadBalancerServer' via
a multiprocessing queue.
Since the queue is shared amongst workers, only one worker will handle
a given connection.
c sH t � j|g|�R i |�� || _t�� | _tj| jd�| _| j� � d S )N)�target)
r- r. r/ � threading�Event�_stop�Thread�socket_queue_thread�thread�start)r1 r/ r� rW r2 r3 r r
r. � s
zLoadBalancerWorker.__init__c s" | j �� | j�� t� �� d S r5 )r� �setr� �joinr- r8 r9 r3 r r
r8 � s
zLoadBalancerWorker.closec C sn z* z | j �dd�\}}W n tjy | j�� r Y W d S Y qw | j�| j||� q t t
fy6 Y d S w �NTr ) r/ r �queue�Emptyr� �is_setr\ r� Z_handle_connection�KeyboardInterrupt�
SystemExit)r1 Z
client_socketrM r r r
r� � s
���
�z&LoadBalancerWorker.socket_queue_thread)r r r
r r. r8 r� rP r r r3 r
r� � s
r� c s* e Zd ZdZd� fdd� Zdd� Z� ZS )�TCPClientKeepAlivezN
Override _create_stream() in TCPClient to enable keep alive support.
Nc s || _ t� j|d� d S )N�r` )r r- r. )r1 r r` r3 r r
r. � s zTCPClientKeepAlive.__init__c K sP t | j�}t|| j� tjjjj||d�}tjjjdk r!|� |�S ||� |�fS )z�
Override _create_stream() in TCPClient.
Tornado 4.5 added the kwargs 'source_ip' and 'source_port'.
Due to this, use **kwargs to swallow these and any future
kwargs to maintain compatibility.
)�max_buffer_size)r� )
r r r* rG rH rI r� ZIOStream� version_inforj )r1 r� �af�addrr2 r) r� r r r
�_create_stream� s
�
z!TCPClientKeepAlive._create_streamr5 )r r r
r r. r� rP r r r3 r
r� � s r� c @ s� e Zd ZdZ d dd�Zdd� Zejjj j
dd� �Z d d
� Z ejjj j
d d � �Z
ejjj j
d
d� �Zejjj j
dd� �Zdd� Zdd� Zdd� Zdd� Zejjj j
d dd��ZdS ) ri z*
Low-level message sending client
Nc
C s� || _ || _|| _|| _| | _|| _|| _|p tjj j
j � � | _
tjj�| j
�� t||d�| _W d � n1 s:w Y i | _d | _d | _d| _d| _tjj j�� | _d| _d | _ |� dd�| _ d S )Nr� FZtcp_reconnect_backoffr ) r �host�portrc rf rd re rG rH rI r� r� r� r\ rm r� r� r� � _tcp_client�send_future_map�_read_until_future�_on_recvr_ �_closed�
concurrent�FutureZ_connecting_future�_stream_return_runningrx r �backoff)
r1 r r� r� r\ r` rd re rc rf r r r
r. s( �zMessageClient.__init__c C s$ | j rd S d| _ | j�d| j� d S r� )r_ r\ � add_timeout� check_closer9 r r r
r8 ? s zMessageClient.closec C s: | j s| j�� d | _d| _d| _d S | j�d| j� d S )NFTr ) r� r� r8 rx r_ r� r\ r� r� r9 r r r
r� E s
zMessageClient.check_closec C r: r5 r; r9 r r r
r< P r= zMessageClient.__del__c
k s� � | j s| jr| j | jd�}d }|d u rg| jsg| jsgz| jjt| jdd�| jfd| j �
d�i|��V }W n( t y\ } z t �
d| j| j|| j� tjjj�| j�V W Y d }~nd }~ww |d u rg| jsg| jrtjjj�|��)N)rc rf Tr r� r� zgTCP Message Client encountered an exception while connecting to %s:%s: %r, will reconnect in %d seconds)rc rf r� r_ r� rj r r� r� r r r� r� �warningr� rG rH rI rs �sleeprt )r1 r2 r� r� r r r
� getstreamU s: � � �
� �� ���zMessageClient.getstreamc c sR � | j d u r#| �� V | _ | j r%| js| j�| j� | jr'| �d� d S d S d S d S ra )rx r� r� r\ r� �_stream_returnrd r9 r r r
rj q s �
�zMessageClient.connectc
c sP � d| _ tjj�� }| j�s#zM| jjddd�V }|�|� |D ]:}tj j
� |�}|d }|d }|� d�}|| j
v rD| j
�|��|� q | jd urS| j�| j||� q t�d|� q W n� tjjjjy� } zJt�d| j| j� | j
� � D ]}|� |� qui | _
| js�| j r�W Y d }~d S | j r�| � � | j} d | _| r�| � � tjj�� }| �!� V W Y d }~nod }~w t"y� d | j#v r�t�$d
� nt%�Y nV t&�y } zItjd dd � | j
� � D ]}|� |� q�i | _
| js�| j r�W Y d }~d S | j r�| � � | j} d | _| �r
| � � tjj�� }| �!� V W Y d }~nd }~ww | jrd
| _ d S )NTr� r� r� rv �midz7Got response for message_id %s that we are not trackingz*tcp stream to %s:%s closed, unable to recv� detect_modez[There was an error trying to use TCP transport; attempting to fallback to another transportzException parsing responser� F)'r� rG rm rn r� r_ rx r� r� rp rq rr r r� r� Z
set_resultr� r\ r� r� �errorrH rI r� r� �debugr� r� �values�
set_exceptionr� re r8 rj � TypeErrorr �infor r� )
r1 r� r� r� r� rv �
message_idrN �futurer� r r r
r� { s� �
��� �
�� ���
Az MessageClient._stream_returnc C s t t�� �S r5 )r� �uuidZuuid4r9 r r r
� _message_id� r= zMessageClient._message_idc s( � du r � | _ dS � fdd�}|| _ dS )zU
Register a callback for received messages (that we didn't initiate)
Nc s � |� d S r5 r )r� rv �r} r r
� wrap_recv� r= z(MessageClient.on_recv.<locals>.wrap_recv)r� )r1 r} r r r� r
r| � s
zMessageClient.on_recvc C s* || j vrd S | j �|�}| j�|� d S r5 )Zsend_timeout_mapr� r\ Zremove_timeout)r1 r� �timeoutr r r
�remove_message_timeout� s
z$MessageClient.remove_message_timeoutc C s8 || j vrd S | j �|�}|d ur|�td�� d S d S )NzMessage timed out)r� r� r� r )r1 r� rz r� r r r
�timeout_message� s
�z MessageClient.timeout_messageFc # s� � �j rt� ���� }d|i}tjjj�� }� d ur&� �fdd�}|�|� |�j |< �j
� d�du r5d}|d urC�j �
|�j||� tjjj||d��tjjjj��fdd ��} �j �| � |V }
tjjj�|
��)
Nr� c s | � � }�j�� |� d S r5 )�resultr\ � add_callback)r� �response)r} r1 r r
�
handle_future� s z)MessageClient.send.<locals>.handle_futurer� Tr r� c 3 s( � �� � V �jr�j�� �V d S d S r5 )rj rx ry r )r� r1 r r
�_do_send� s
�
�z$MessageClient.send.<locals>._do_send)r_ r r� rG rH rI r� r� Zadd_done_callbackr� r r r\ Z
call_laterr rp rq r� rs r� r rt ) r1 rz r r} �rawr� r� r� r r �recvr )r} r� r1 r
r{ � s( �
zMessageClient.send)NNNNNN)NNF)r r r
r r. r8 rG rH rI rs r� r� r< r� rj r� r� r| r r r{ r r r r
ri s2
�$
F
ri c @ s( e Zd ZdZdd� Zdd� Zdd� ZdS ) �
Subscriberz=
Client object for use with the TCP publisher server
c C s" || _ || _d| _d | _d | _d S r[ )r� rM r_ r� �id_)r1 r� rM r r r
r. s
zSubscriber.__init__c C sR | j rd S d| _ | j�� s#| j�� | jd ur%| j�� r'| j�� d S d S d S d S ra )r_ r� �closedr8 r� Zdoner� r9 r r r
r8 s
�zSubscriber.closec C r: r5 r; r9 r r r
r< % r= zSubscriber.__del__N)r r r
r r. r8 r< r r r r
r s
r c sf e Zd ZdZ d� fdd� Zdd� Zdd� Zejj j
j d d
� �Z d d � Z
ejj j
j dd
d��Z� ZS )� PubServerz
TCP publisher
Nc sf t � j|�d�d� || _|| _d| _t� | _d| _|r || _ ndd� | _ |r,|| _
d S dd� | _
d S )Nr� r� Fc S r� r5 r )�
subscriberrz r r r
�<lambda>< � z$PubServer.__init__.<locals>.<lambda>c S r~ r5 r )r r r r
r @ r ) r- r. r r\ r r_ r� r� Zpresence_events�presence_callback�remove_presence_callback)r1 r r\ r r r3 r r
r. 0 s
zPubServer.__init__c C s* | j rd S d| _ | jD ]}|j�� q d S ra )r_ r� r� Z
disconnectr� r r r
r8 B s
�zPubServer.closec C r: r5 r; r9 r r r
r< J r= zPubServer.__del__c
c s � t jj�� }| js�z-|jjddd�|_|jV }|�|� |D ]}t j j
� |�}|d }| j r5| � ||� q W nG t j
jjjyd } z t�d|j� |�� | �|� | j�|� W Y d }~d S d }~w ty~ } ztjd|jdd� W Y d }~qd }~ww | jr
d S d S )Nr� Tr� rv z'tcp stream to %s closed, unable to recvz"Exception parsing response from %sr� )rG rm rn r� r_ r� r� r� r� rp rq rr r rH rI r� r� r� r� rM r8 r r� �discardr� r� )r1 r� r� r� r� rv rN r r r
� _stream_readO s: �
��
�� ���zPubServer._stream_readc C s6 t �d|� t||�}| j�|� | j�| j|� d S )NzSubscriber at %s connected)r� r� r r� �addr\ r� r )r1 r� rM r� r r r
r� h s
zPubServer.handle_streamc
c s" � t �d||� tjj�|�}g }|rN|D ]7}d}| jD ]%}||jkrAz |j� |�V d}W q tj
j j j
y@ |�|� Y q w q |sLt �d|| j� qn!| jD ] }z |j� |�V W qQ tj
j j j
yn |�|� Y qQw |D ]}t �d|j� |�� | �|� | j�|� qqt �d� d S )Nz'TCP PubServer sending payload: %s
%rFTz"Publish target %s not connected %rz0Subscriber at %s has disconnected from publisherz)TCP PubServer finished publishing payload)r� r� rG rp rq r� r� r r� ry rH rI r� r� r� r� rM r8 r r )r1 �packageZ
topic_listr� � to_removeZtopic�sentr� r r r
�publish_payloado sD �
����
��
zPubServer.publish_payload)NNNr5 )r r r
r r. r8 r< rG rH rI rs r� r r� r rP r r r3 r
r + s �
r c @ sr e Zd ZdZdZdd� Zedd� �Zdd� Zd d
� Z dd d
�Z
dd� Z e j
jjjdd� �Zdd� Zdd� Zd S )�TCPPublishServerz)
Tornado based TCP PublishServer
r, c C s || _ d | _d S r5 )r �pub_sockr� r r r
r. � s
zTCPPublishServer.__init__c C s | j �dd� S )NZ
order_mastersF)r r r9 r r r
�
topic_support� s z TCPPublishServer.topic_supportc C s | � |d � d S �Nr )r. )r1 �stater r r
� __setstate__� s z TCPPublishServer.__setstate__c C s
d| j iS r )r r9 r r r
� __getstate__� s
z TCPPublishServer.__getstate__Nc C sp t jjj�� }|�� || _t| j|||d� | _ }t
| j�}|� t j
t jd� t|| j� |�d� |�t| jd�� |�| j� |�|� | j�dd�dkrZt| j�dd ��}n
tj�| jd
d �}|| _ t jj j |||d �}t � d
|� t j j!�"d�� |�#� W d � n1 s�w Y z z|�#� W n t$t%fy� Y nw W |�&� dS W |�&� dS |�&� w )zK
Bind to the interface specified in the configuration file
)r\ r r r r rh �ipc_mode� rZ �tcp_master_publish_pull� �sock_dir�publish_pull.ipc)r\ Zpayload_handlerz Starting the Salt Puller on %s� N)'rG rH rI r� r� Z make_currentr\ r r �
pub_serverr r& r r' r? r* r@ rA r rB rC r� r r �os�pathr� rp �ipcZIPCMessageServerr� r� rm �filesZ set_umaskr� r� r� r8 ) r1 r r r r\ r) r) �pull_uriZ pull_sockr r r
�publish_daemon� sL �
�
� �� �z TCPPublishServer.publish_daemonc C s |j | j| jjd� dS )z�
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
)r� N)r� r/ r4 r r� r r r
r� � s zTCPPublishServer.pre_forkc g s( � | j j|g|�R � V }tjjj�|��r5 )r) r rG rH rI rs rt )r1 r� rW �retr r r
r � s �z TCPPublishServer.publish_payloadc K sx | j �dd�dkrt| j �dd��}n
tj�| j d d�}| js4tjj j
tj j j
|fdd �| _| j�� | j�|� d
S ) z+
Publish "load" to minions
r" r# rZ r$ r% r& r' r\ )Z
loop_kwargN)r r r r* r+ r� r rG rm r� Z SyncWrapperrp r, ZIPCMessageClientrj r{ )r1 r� r2 r. r r r
�publish� s �
zTCPPublishServer.publishc C s | j r
| j �� d | _ d S d S r5 )r r8 r9 r r r
r8 � s
�zTCPPublishServer.closer� )r r r
r rC r. r� r r r! r/ r� rG rH rI rs r� r r1 r8 r r r r
r � s
�6
r c sV e Zd ZdZdZ� fdd�Zejjj j
dd� �Z ejjj j
d dd ��Z d
d � Z
� ZS )
� TCPReqClientz)
Tornado based TCP RequestClient
rZ c
s� t � j||fi |�� || _|| _tj�| jd �}|j�dd�\}}|t |�f}|�
d�}t j j
j||t |�|||�
d�|�
d�d�| _d| _d S ) NZ
master_uri�:r r` rc Zsource_ret_port)r\ r` rc rf F)r- r. r r\ �urllib�parse�urlparse�netloc�rsplitr r rG rp rZ ri r] r_ ) r1 r r\ r2 r5 Z master_hostZ master_portZ master_addrr` r3 r r
r. s"
�
zTCPReqClient.__init__c c s � d| _ | j�� V d S ra )rg r] rj r9 r r r
rj s �zTCPReqClient.connect�< c c s$ � | j j||d�V }tjjj�|��)N)r )r] r{ rG rH rI rs rt )r1 �loadr r0 r r r
r{ $ s �zTCPReqClient.sendc C s | j rd S d| _ | j�� d S ra rb r9 r r r
r8 ) s zTCPReqClient.close)r9 )r r r
r r� r. rG rH rI rs r� rj r{ r8 rP r r r3 r
r2 s
r2 )Gr rK �loggingr� r* r� r r� r4 r� Zsalt.ext.tornadorG Zsalt.ext.tornado.concurrentZsalt.ext.tornado.genZsalt.ext.tornado.iostreamZsalt.ext.tornado.netutilZsalt.ext.tornado.tcpclientZsalt.ext.tornado.tcpserverZ salt.masterZ salt.payloadZsalt.transport.frameZsalt.transport.ipcZsalt.utils.asynchronousZsalt.utils.filesZsalt.utils.msgpackZsalt.utils.platformZsalt.utils.versionsZsalt.exceptionsr r Zsalt.utils.networkr Zsalt.utils.processr rm r� r� r� Zsalt.ext.tornado.util� getLoggerr r� r� r r r r* r+ rQ rp �baseZ
PublishClientrY ZDaemonizedRequestServerr� rH rI Z tcpserverZ TCPServerr� r� Z tcpclientZ TCPClientr� ri r r ZDaemonizedPublishServerr Z
RequestClientr2 r r r r
�<module> sh
><ErG'" t"hq