| Server IP : 198.38.94.13 / Your IP : 216.73.217.33 Web Server : Apache System : Linux d4744.dxb1.stableserver.net 5.14.0-611.49.1.el9_7.x86_64 #1 SMP PREEMPT_DYNAMIC Tue Apr 21 16:39:08 EDT 2026 x86_64 User : revivere ( 1140) PHP Version : 8.2.33 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : OFF | Pkexec : OFF Directory : /opt/saltstack/salt/lib/python3.10/site-packages/zmq/devices/__pycache__/ |
Upload File : |
o
;jF% � @ s� d Z ddlZddlmZ ddlmZ ddlmZmZm Z m
Z
m Z ddl Z ddl m
Z
mZmZmZmZmZmZmZ G dd� d�ZG d d
� d
e�ZG d d � d e�ZG d
d� de�Zg d�ZdS )z2Classes for running 0MQ Devices in the background.� N)�Process)�Thread)�Any�Callable�List�Optional�Tuple)�ENOTSOCK�ETERM�PUSH�QUEUE�Context� ZMQBindError�ZMQError�proxyc @ s� e Zd ZU dZejZeg ejf e d< e
e d< e e d< e e d< e e d< e e
e d< e e
e d< e ee ef e d < e e
e d
< e e
e d < e ee ef e d < e e
e d
< e ej e d< eddfde dee dee ddfdd�Zde
ddfdd�Zde
de fdd�Zde
ddfdd�Zde deddfd d �Zde
ddfd d �Zde
de fd d!�Zde
fd"d#�Zde defd$d%�Zde
de fd&d'�Z deejejf fd(d)�Z d5d*d+�Z d,d-� Z d5d.d/�Z d5d0d1�Z!d6d2ee" ddfd3d4�Z#dS )7�Deviceaa A 0MQ Device to be run in the background.
You do not pass Socket instances to this, but rather Socket types::
Device(device_type, in_socket_type, out_socket_type)
For instance::
dev = Device(zmq.QUEUE, zmq.DEALER, zmq.ROUTER)
Similar to zmq.device, but socket types instead of sockets themselves are
passed, and the sockets are created in the work thread, to avoid issues
with thread safety. As a result, additional bind_{in|out} and
connect_{in|out} methods and setsockopt_{in|out} allow users to specify
connections for the sockets.
Parameters
----------
device_type : int
The 0MQ Device type
{in|out}_type : int
zmq socket types, to be passed later to context.socket(). e.g.
zmq.PUB, zmq.SUB, zmq.REQ. If out_type is < 0, then in_socket is used
for both in_socket and out_socket.
Methods
-------
bind_{in_out}(iface)
passthrough for ``{in|out}_socket.bind(iface)``, to be called in the thread
connect_{in_out}(iface)
passthrough for ``{in|out}_socket.connect(iface)``, to be called in the
thread
setsockopt_{in_out}(opt,value)
passthrough for ``{in|out}_socket.setsockopt(opt, value)``, to be called in
the thread
Attributes
----------
daemon : bool
sets whether the thread should be run as a daemon
Default is true, because if it is false, the thread will not
exit unless it is killed
context_factory : callable
This is a class attribute.
Function for creating the Context. This will be Context.instance
in ThreadDevices, and Context in ProcessDevices. The only reason
it is not instance() in ProcessDevices is that there may be a stale
Context instance already initialized, and the forked environment
should *never* try to use it.
�context_factory�daemon� device_type�in_type�out_type� _in_binds� _in_connects� _in_sockopts�
_out_binds�
_out_connects�
_out_sockopts�
_random_addrs�_socketsN�returnc C sr || _ |d u r td��|d u rtd��|| _|| _g | _g | _g | _g | _g | _g | _ g | _
d| _ d| _ g | _
d S )Nzin_type must be specifiedzout_type must be specifiedTF)r � TypeErrorr r r r r r r r r r �doner )�selfr r r � r# �J/opt/saltstack/salt/lib/python3.10/site-packages/zmq/devices/basedevice.py�__init__V s"
zDevice.__init__�addrc C � | j �|� dS )z`Enqueue ZMQ address for binding on in_socket.
See zmq.Socket.bind for details.
N)r �append�r" r& r# r# r$ �bind_inn � zDevice.bind_inc O �0 | j |g|�R i |��}| �|� d|� �� |S )z�Enqueue a random port on the given interface for binding on
in_socket.
See zmq.Socket.bind_to_random_port for details.
.. versionadded:: 18.0
�:)�_reserve_random_portr* �r" r& �args�kwargs�portr# r# r$ �bind_in_to_random_portu � z Device.bind_in_to_random_portc C r' )zfEnqueue ZMQ address for connecting on in_socket.
See zmq.Socket.connect for details.
N)r r( r) r# r# r$ �
connect_in� r+ zDevice.connect_in�opt�valuec C � | j �||f� dS )zeEnqueue setsockopt(opt, value) for in_socket
See zmq.Socket.setsockopt for details.
N)r r( �r" r6 r7 r# r# r$ �
setsockopt_in� � zDevice.setsockopt_inc C r' )zaEnqueue ZMQ address for binding on out_socket.
See zmq.Socket.bind for details.
N)r r( r) r# r# r$ �bind_out� r+ zDevice.bind_outc O r, )z�Enqueue a random port on the given interface for binding on
out_socket.
See zmq.Socket.bind_to_random_port for details.
.. versionadded:: 18.0
r- )r. r< r/ r# r# r$ �bind_out_to_random_port� r4 z Device.bind_out_to_random_portc C r' )zgEnqueue ZMQ address for connecting on out_socket.
See zmq.Socket.connect for details.
N)r r( r) r# r# r$ � connect_out� r+ zDevice.connect_outc C r8 )zfEnqueue setsockopt(opt, value) for out_socket
See zmq.Socket.setsockopt for details.
N)r r( r9 r# r# r$ �setsockopt_out� r; zDevice.setsockopt_outc O s� t � �Q}|�t��3}td�D ] }|j|g|�R i |��}|� d|� �}|| jv r)q ntd��| j�|� W d � n1 s?w Y W d � |S W d � |S 1 sWw Y |S )N� r- z Could not reserve random port.)r
�socketr �rangeZbind_to_random_portr r r( ) r" r& r0 r1 �ctxZbinder�ir2 �new_addrr# r# r$ r. � s$
�
��
��zDevice._reserve_random_portc C s� | � � }|| _|�| j�}| j�|� | jdk r|}n |�| j�}| j�|� | jD ]
\}}|�||� q*| j D ]
\}}|�||� q8| j
D ]}|� |� qF| j D ]}|� |� qQ| j
D ]}|�|� q\| jD ]}|�|� qg||fS )Nr )r �_contextrA r r r( r r �
setsockoptr r �bindr r �connectr )r" rC �ins�outsr6 r7 Zifacer# r# r$ �_setup_sockets� s*
zDevice._setup_socketsc C s | � � \}}t||� dS )zpThe runner method.
Do not call me directly, instead call ``self.start()``, just like a Thread.
N)rL r )r" rJ rK r# r# r$ �
run_device� s zDevice.run_devicec C s" | j D ] }|r|js|�� qdS )zCleanup sockets we createdN)r �closed�close)r" �sr# r# r$ �_close_sockets� s
��zDevice._close_socketsc
C s~ z6z| � � W n ty" } z|jtthv rn� W Y d}~nd}~ww W d| _| �� dS W d| _| �� dS d| _| �� w )z"wrap run_device in try/catch ETERMNT)rM r �errnor
r r! rQ )r" �er# r# r$ �run� s" ���� � �
z
Device.runc C s | � � S )z>Start the device. Override me in subclass for other launchers.)rT �r" r# r# r$ �start s z Device.start�timeoutc C sb t �� }|}| js+|dur|| |ks/t �d� t �� }| js-|dur|| |krdS dS dS dS )z\wait for me to finish, like Thread.join.
Reimplemented appropriately by subclasses.Ng����MbP?)�time� monotonicr! �sleep)r" rW ZticZtocr# r# r$ �join s
*�z Device.join�r N�N)$�__name__�
__module__� __qualname__�__doc__r
�instancer r �zmq�__annotations__�bool�intr �strr r ZSocketr r r% r* r3 r5 r: r< r= r> r? r. rL rM rQ rT rV �floatr[ r# r# r# r$ r sV
3 ����
�
r c @ sL e Zd ZU dZdZeed< dZeed< d dd�Zd de e
ddfd d
�Z dS )
�BackgroundDevicezEBase class for launching Devices in background processes and threads.N�launcher�
_launch_classr c C s$ | j | jd�| _| j| j_| j�� S )N)�target)rk rT rj r rV rU r# r# r$ rV s
zBackgroundDevice.startrW c C s | j j|d�S )N)rW )rj r[ )r" rW r# r# r$ r[ s zBackgroundDevice.joinr\ r] ) r^ r_ r` ra rj r rd rk rV r rh r[ r# r# r# r$ ri s
ri c @ s e Zd ZdZeZdS )� ThreadDevicezSA Device that will be run in a background Thread.
See Device for details.
N)r^ r_ r` ra r rk r# r# r# r$ rm s rm c @ s e Zd ZdZeZeZdS )�
ProcessDevicezTA Device that will be run in a background Process.
See Device for details.
N)r^ r_ r` ra r rk r
r r# r# r# r$ rn ) s
rn )r rm rn )ra rX �multiprocessingr � threadingr �typingr r r r r rc r r
r r r
r r r r ri rm rn �__all__r# r# r# r$ �<module> s (