File: //usr/local/lib/python3.9/site-packages/kombu/transport/__pycache__/filesystem.cpython-39.pyc
a
X>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 �D/usr/local/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: �queuesr 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' rJ rK �filename�fr r r rI � 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)rR �listdir�data_folder_in�sorted�len�pop�find�store_processed�processed_folder�tempfile�
gettempdir�shutil�moverS rT r6 r2 r3 r4 �remover r r r )r7 r' �
queue_find�folderrV ra rW rJ 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 rX r )
rR rZ r[ r] r^ r_ rS rT rf r6 �r7 r' �countrg rh rV 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 rX r )rR rZ r[ r] r^ r_ rj r r r �_size" s
z
Channel._sizec C s
| j jjS rH )�
connection�client�transport_options�r7 r r r rp 3 s zChannel.transport_optionsc C s | j �dd�S )Nr[ Zdata_in�rp �getrq r r r r[ 7 s zChannel.data_folder_inc C s | j �dd�S )NrU Zdata_outrr rq r r r rU ; s zChannel.data_folder_outc C s | j �dd�S )Nr` Frr rq r r r r` ? s zChannel.store_processedc C s | j �dd�S )Nra � processedrr rq r r r ra C s zChannel.processed_folderc C s t | j�dd��S )Nr1 �control)r rp rs rq r r r r1 G s zChannel.control_folderN)�__name__�
__module__� __qualname__�__doc__Zsupports_fanoutr; rG rL rI ri rl rm �propertyrp r
r[ rU r` ra 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 rH )�super�__init__� global_state�state)r7 ro rK �� __class__r r r [ s zTransport.__init__c C s dS )NzN/Ar rq r r r �driver_version_ s zTransport.driver_version)rv rw rx ry r r{ Z
implements�extend� frozensetr( Z BrokerStater� � default_portZ driver_typeZ driver_namer r� �
__classcell__r r r� r r{ L s
� r{ )2ry �
__future__r rR rd rb rP � collectionsr �pathlibr r' r �timer Zkombu.exceptionsr Zkombu.transportr Zkombu.utils.encodingr r
Zkombu.utils.jsonr r Zkombu.utils.objectsr
�VERSIONrT �map�str� __version__�nameZ
pywintypesZwin32conr ZLOCKFILE_EXCLUSIVE_LOCKr r! ZLOCKFILE_FAIL_IMMEDIATELYZLOCK_NBZ
OVERLAPPEDr r r r" � RuntimeErrorr$ r( r{ r r r r �<module> sP [
�� .