o
    j                     @   s   d Z zddlmZ W n ey   ddlmZ Y nw ddlmZ ddlmZ ddlm	Z	 ddl
mZ ddlmZ dd	lmZ dd
lmZmZ G dd deZdS )
    )Queue)StringIO)linesep   ) decode_persistent_search_control)AsyncStrategy)LDAPLDIFError)prepare_for_stream)"persistent_search_response_to_ldifadd_ldif_headerc                   @   s@   e Zd ZdZdd Zdd Zdd Zdd	 Zd
d Zdd Z	dS )AsyncStreamStrategyz|
    This strategy is asynchronous. It streams responses in a generator as they appear in the self._responses container
    c                 C   sn   t | | d| _t| _d| _d | _t | _d| _	d | _
d| _d | _|jr/t|j| _| `d S t | _| `d S )NTF)r   __init__
can_streamr   line_separator
all_base64streamdictorder_header_addedpersistent_search_message_id	streamingcallback	pool_sizer   events	_requests)selfldap_connection r   /root/aizidognhua/tmp/workspace/projects/ec89d86c-575f-41c9-af57-ac45cbdbf775/venv/lib/python3.10/site-packages/ldap3/strategy/asyncStream.pyr   /   s   zAsyncStreamStrategy.__init__c                 C   sF   t |  | jr| jrt| jtr| jjr!| t  d S d S d S d S N)r   _start_listenr   r   
isinstancer   closed
set_streamr   r   r   r   r!   A   s   
z!AsyncStreamStrategy._start_listenc                 C   s"   t |  | jr| j  d S d S r    )r   _stop_listenr   r   closer%   r   r   r   r&   G   s   
z AsyncStreamStrategy._stop_listenc                 C   s\  || j kr| j g | j|< W d    n1 sw   Y  | jr| js?| j dkr?tdgd }| jt	|| j
 | j
  t|}| jr|r| jjs| j
|}| jst| j dkrtd| _tdgd }| jt	|| j
 | j
  | jt	|| j
 | j
  d S d S d S d S t|}|r|| |d d= | js| j| d S | | d S d S )Nr   -Tcontrolsz2.16.840.1.113730.3.4.7)r   
async_lock
_responsesr   r   r   tellr   writer
   r   r   
connectionr#   joinr   updater   r   put)r   
message_idchangeheader
ldif_linesfragmentnotificationr   r   r   accumulate_streamL   s2   
 

z%AsyncStreamStrategy.accumulate_streamc                 C   s   | j r| jS d S r    )r   r   r%   r   r   r   
get_streamf   s   zAsyncStreamStrategy.get_streamc              	   C   sL   d}z|  s	d}W n ttfy   d}Y nw |rtd|| _d| _d S )NFTzstream must be writable)writable
ValueErrorAttributeErrorr	   r   r   )r   valueerrorr   r   r   r$   k   s   
zAsyncStreamStrategy.set_streamN)
__name__
__module____qualname____doc__r   r!   r&   r8   r9   r$   r   r   r   r   r   +   s    r   N)rB   queuer   ImportErrorior   osr   protocol.rfc2849r   strategy.asynchronousr   core.exceptionsr	   
utils.convr
   r   r   r   r   r   r   r   <module>   s    