Файловый менеджер - Редактировать - /opt/imh-python/lib/python3.9/site-packages/kombu/transport/__pycache__/filesystem.cpython-39.pyc
Ðазад
a o�h�( � @ sz d Z ddlmZ ddlZddlZddlZddlZddlmZ ddl m Z ddl m Z ddl mZ ddlmZ dd lmZ dd lmZmZ dd lmZmZ dd lmZ d Zd� e e e��Z ej dk�rddl!Z!ddl"Z"ddl#Z#e"j$Z%dZ&e"j'Z(e!�)� Z*dd� Z+dd� Z,n>ej dk�r<ddl-Z-ddl-m%Z%m&Z& dd� Z+dd� Z,ne.d��edg d��Z/G dd � d ej0�Z0G d d � d ej1�Z1dS ) a� File-system Transport module for kombu. Transport using the file-system as the message store. Messages written to the queue are stored in `data_folder_in` directory and messages read from the queue are read from `data_folder_out` directory. Both directories must be created manually. Simple example: * Producer: .. code-block:: python import kombu conn = kombu.Connection( 'filesystem://', transport_options={ 'data_folder_in': 'data_in', 'data_folder_out': 'data_out' } ) conn.connect() test_queue = kombu.Queue('test', routing_key='test') with conn as conn: with conn.default_channel as channel: producer = kombu.Producer(channel) producer.publish( {'hello': 'world'}, retry=True, exchange=test_queue.exchange, routing_key=test_queue.routing_key, declare=[test_queue], serializer='pickle' ) * Consumer: .. code-block:: python import kombu conn = kombu.Connection( 'filesystem://', transport_options={ 'data_folder_in': 'data_out', 'data_folder_out': 'data_in' } ) conn.connect() def callback(body, message): print(body, message) message.ack() test_queue = kombu.Queue('test', routing_key='test') with conn as conn: with conn.default_channel as channel: consumer = kombu.Consumer( conn, [test_queue], accept=['pickle'] ) consumer.register_callback(callback) with consumer: conn.drain_events(timeout=1) Features ======== * Type: Virtual * Supports Direct: Yes * Supports Topic: Yes * Supports Fanout: Yes * Supports Priority: No * Supports TTL: No Connection String ================= Connection string is in the following format: .. code-block:: filesystem:// Transport Options ================= * ``data_folder_in`` - directory where are messages stored when written to queue. * ``data_folder_out`` - directory from which are messages read when read from queue. * ``store_processed`` - if set to True, all processed messages are backed up to ``processed_folder``. * ``processed_folder`` - directory where are backed up processed files. * ``control_folder`` - directory where are exchange-queue table stored. � )� annotationsN)� namedtuple)�Path)�Empty)� monotonic)� ChannelError)�virtual)� bytes_to_str� str_to_bytes)�dumps�loads)�cached_property)� r r �.�ntc C s$ t �| �� �}t �||ddt� dS )�Create file lock.r � � N)� win32file�_get_osfhandle�filenoZ LockFileEx� __overlapped)�file�flags�hfile� r ��/root/rpmbuild/BUILDROOT/imh-python39-modules-3.9.7-92.el8.x86_64/opt/imh-python/lib/python3.9/site-packages/kombu/transport/filesystem.py�lock} s r c C s" t �| �� �}t �|ddt� dS )�Remove file lock.r r N)r r r Z UnlockFileExr )r r r r r �unlock� s r �posix)�LOCK_EX�LOCK_SHc C s t �| �� |� dS )r N)�fcntl�flockr )r r r r r r � s c C s t �| �� t j� dS )r N)r" r# r ZLOCK_UN)r r r r r � s z9Filesystem plugin only defined for NT and POSIX platforms�exchange_queue_t)� routing_key�pattern�queuec @ s� e Zd ZdZdZdd� Zdd� Zdd� Zd d � Zd d � Z d d� Z dd� Z e dd� �Z edd� �Zedd� �Zedd� �Zedd� �Ze dd � �Zd S ) �ChannelzFilesystem Channel.Tc C s� | j |� d� }z^|�d�}z<t|t� tt|�� ��}dd� |D �W t|� |�� W S t|� |�� 0 W n4 t y� g Y S t y� t d|� ���Y n0 d S )N� .exchange�rc S s g | ] }t |� �qS r �r$ ��.0�qr r r � <listcomp>� � z%Channel.get_table.<locals>.<listcomp>z Cannot open ) �control_folder�openr r! r r �readr �close�FileNotFoundError�OSErrorr )�self�exchanger �f_obj�exchange_tabler r r � get_table� s$ � � zChannel.get_tablec C s | j |� d� }| j jdd� t|p&d|p,d|p2d�}z�|�� r�|jddd�}t|t� tt|� � ��}dd � |D �} || vr�| � d|� |� d� |� t t| ��� n0|jd dd�}t|t� |g} |� t t| ��� W t|� |�� nt|� |�� 0 d S ) Nr) T)�exist_ok� zrb+r �� bufferingc S s g | ] }t |� �qS r r+ r, r r r r/ � r0 z'Channel._queue_bind.<locals>.<listcomp>�wb)r1 �mkdirr$ �existsr2 r r r r r3 �insert�seek�writer r r r4 ) r7 r8 r% r& r' r Z queue_valr9 r: Zqueuesr r r � _queue_bind� s. � �zChannel._queue_bindc K s* | � |�D ]}| j|j|fi |�� q d S �N)r; �_putr' )r7 r8 �payloadr% �kwargsr. r r r � _put_fanout� s zChannel._put_fanoutc K s� d� ttt� d ��t�� |�}tj�| j |�}zfz.t |ddd�}t |t � |� tt|��� W n" ty� td|�d���Y n0 W t|� |�� nt|� |�� 0 dS ) zPut `message` onto `queue`.z {}_{}.{}.msgi� r@ r r> zCannot add file z to directoryN)�format�int�roundr �uuid�uuid4�os�path�join�data_folder_outr2 r r rE r r r6 r r r4 )r7 r' rI rJ �filename�fr r r rH � s � � �z Channel._putc C s d| d }t �| j�}t|�}t|�dkr�|�d�}|�|�dk rFq | jrT| j}nt � � }z t � t j �| j|�|� W n ty� Y q Y n0 t j �||�}z.t|d�}|�� }|�� | js�t �|� W n" ty� td|�d���Y n0 tt|��S t� �dS )z Get next message from `queue`.r �.msgr �rbzCannot read file z from queue.N)rQ �listdir�data_folder_in�sorted�len�pop�find�store_processed�processed_folder�tempfile� gettempdir�shutil�moverR rS r6 r2 r3 r4 �remover r r r )r7 r' � queue_find�folderrU r` rV rI r r r �_get� s: � � z Channel._getc C s� d}d| d }t �| j�}t|�dkr~|�� }z8|�|�dk rDW q t j�| j|�}t �|� |d7 }W q t yz Y q 0 q |S )z!Remove all messages from `queue`.r r rW r ) rQ rY rZ r\ r] r^ rR rS re r6 �r7 r' �countrf rg rU r r r �_purge s zChannel._purgec C sN d}d|� d�}t �| j�}t|�dkrJ|�� }|�|�dk r@q |d7 }q |S )z<Return the number of messages in `queue` as an :class:`int`.r r rW r )rQ rY rZ r\ r] r^ ri r r r �_size" s z Channel._sizec C s | j jjS rG )� connection�client�transport_options�r7 r r r ro 3 s zChannel.transport_optionsc C s | j �dd�S )NrZ Zdata_in�ro �getrp r r r rZ 7 s zChannel.data_folder_inc C s | j �dd�S )NrT Zdata_outrq rp r r r rT ; s zChannel.data_folder_outc C s | j �dd�S )Nr_ Frq rp r r r r_ ? s zChannel.store_processedc C s | j �dd�S )Nr` � processedrq rp r r r r` C s zChannel.processed_folderc C s t | j�dd��S )Nr1 �control)r ro rr rp r r r r1 G s zChannel.control_folderN)�__name__� __module__� __qualname__�__doc__Zsupports_fanoutr; rF rK rH rh rk rl �propertyro r rZ rT r_ r` r1 r r r r r( � s* ( r( c sZ e Zd ZdZejjjdeg d��d�Ze Z e� � Z dZ dZ dZ� fdd�Zd d � Z� ZS ) � TransportzFilesystem Transport.F)�directZtopicZfanout)Z asynchronousZ exchange_typer � filesystemc s t � j|fi |�� | j| _d S rG )�super�__init__� global_state�state)r7 rn rJ �� __class__r r r~ [ s zTransport.__init__c C s dS )NzN/Ar rp r r r �driver_version_ s zTransport.driver_version)ru rv rw rx r rz Z implements�extend� frozensetr( Z BrokerStater � default_portZ driver_typeZ driver_namer~ r� � __classcell__r r r� r rz L s � rz )2rx � __future__r rQ rc ra rO � collectionsr �pathlibr r' r �timer Zkombu.exceptionsr Zkombu.transportr Zkombu.utils.encodingr r Zkombu.utils.jsonr r Zkombu.utils.objectsr �VERSIONrS �map�str� __version__�nameZ pywintypesZwin32conr ZLOCKFILE_EXCLUSIVE_LOCKr r! ZLOCKFILE_FAIL_IMMEDIATELYZLOCK_NBZ OVERLAPPEDr r r r" � RuntimeErrorr$ r( rz r r r r �<module> sP [ �� .
| ver. 1.1 | |
.
| PHP 8.3.30 | Ð“ÐµÐ½ÐµÑ€Ð°Ñ†Ð¸Ñ Ñтраницы: 0 |
proxy
|
phpinfo
|
ÐаÑтройка