File: //proc/self/root/usr/lib/python3/dist-packages/twisted/protocols/__pycache__/pcp.cpython-38.pyc
U
W[� � @ sL d Z ddlmZ ddlmZ eejej�G dd� d��ZG dd� de�ZdS ) z
Producer-Consumer Proxy.
� )� implementer)�
interfacesc @ st e Zd ZdZdZdZdZdZdZdZ dZ
dd� Z dd� Z d d
� Z
d d � Zd
d� Zdd� Zdd� Zdd� Zdd� ZdS )�BasicProducerConsumerProxyaa
I can act as a man in the middle between any Producer and Consumer.
@ivar producer: the Producer I subscribe to.
@type producer: L{IProducer<interfaces.IProducer>}
@ivar consumer: the Consumer I publish to.
@type consumer: L{IConsumer<interfaces.IConsumer>}
@ivar paused: As a Producer, am I paused?
@type paused: bool
NTFc C s& g | _ |d k r"|| _|�| | j� d S �N)�_buffer�consumer�registerProducer� iAmStreaming)�selfr � r �7/usr/lib/python3/dist-packages/twisted/protocols/pcp.py�__init__" s z#BasicProducerConsumerProxy.__init__c C s d| _ | jr| j�� d S �NT)�paused�producer�pauseProducing�r
r r r r * s z)BasicProducerConsumerProxy.pauseProducingc C sT d| _ | jr0| j�d�| j�� g | jd d �< n | js<d| _| jd k rP| j�� d S )NF� T) r r r �write�joinr �outstandingPullr �resumeProducingr r r r r / s
z*BasicProducerConsumerProxy.resumeProducingc C s& | j d k r| j �� | jd k r"| `d S r )r �
stopProducingr r r r r r < s
z(BasicProducerConsumerProxy.stopProducingc C s@ | j s| js | js | j�|� n | jd k r<| j�|� d| _d S �NF)r r r r �appendr r �r
�datar r r r D s
z BasicProducerConsumerProxy.writec C s | j d k r| j �� | �� d S r )r �finish�unregisterProducerr r r r r M s
z!BasicProducerConsumerProxy.finishc C s || _ || _d S r )r �producerIsStreaming�r
r Z streamingr r r r R s z+BasicProducerConsumerProxy.registerProducerc C s& | j d k r| ` | `| jr"| j�� d S r )r r r r r r r r r V s
z-BasicProducerConsumerProxy.unregisterProducerc C s d| j t| �| jf S )Nz<%s@%x around %s>)� __class__�idr r r r r �__repr__] s z#BasicProducerConsumerProxy.__repr__)�__name__�
__module__� __qualname__�__doc__r r r r r r Zstoppedr
r r r r r r r r# r r r r r s"
r c @ sL e Zd ZdZdZdZdZdd� Zdd� Zdd � Z d
d � Z
d d
� Z dd� Z dS )�ProducerConsumerProxyz�ProducerConsumerProxy with a finite buffer.
When my buffer fills up, I have my parent Producer pause until my buffer
has room in it again.
i Fc C s
d| _ d S r )r r r r r r m s z$ProducerConsumerProxy.pauseProducingc C s� d| _ | jrjd�| j�}| �|�}|t|�k rZ||d � }| jrHtd��|g| jd d �< qng | jd d �< nd}| jr�|r�| js�| jd k r�| j� � | js�| | _
| j d k r�t dd� | jD ��}| j
r�|| jk r�d| _
| j �� n| j
r�| j �� d S )NFr �.Streaming producer did not write all its data.r c S s g | ] }t |��qS r ��len��.0�sr r r �
<listcomp>� s z9ProducerConsumerProxy.resumeProducing.<locals>.<listcomp>)r r r �_writeSomeDatar+ r �AssertionError� unregisteredr r r r �sum�producerPaused�
bufferSizer )r
r � bytesSentZunsent�
bytesBufferedr r r r r s2
��
z%ProducerConsumerProxy.resumeProducingc C s� | j s| js | js | j�|� nV| jd k rv| jr8td��| �|�}d| _|t|�ksv| jrbtd��| j�||d � � | j d k r�| j
r�t dd� | jD ��}|| j kr�| j �
� d| _d S )Nz9Writing fresh data to consumer before my buffer is empty!Fr) c S s g | ] }t |��qS r r* r, r r r r/ � s z/ProducerConsumerProxy.write.<locals>.<listcomp>T)r r r r r r r1 r0 r+ r r r3 r5 r r4 )r
r r6 r7 r r r r � s$
�
�
zProducerConsumerProxy.writec C s$ d| _ t�| ||� |s |�� d S r )r2 r r r r r r r r � s z&ProducerConsumerProxy.registerProducerc C s2 | j d k r| ` | `d| _| jr.| js.| j�� d S r )r r r2 r r r r r r r r � s
z(ProducerConsumerProxy.unregisterProducerc C s" | j dkrdS | j �|� t|�S )z`Write as much of this data as possible.
@returns: The number of bytes written.
Nr )r r r+ r r r r r0 � s
z$ProducerConsumerProxy._writeSomeDataN)
r$ r% r& r' r5 r4 r2 r r r r r r0 r r r r r( a s (r( N) r' Zzope.interfacer Ztwisted.internetr Z IProducerZ IConsumerr r( r r r r �<module> s
R