o
    ¯b  ã                   @   sL   d Z ddlmZ ddlmZ eejejƒG dd„ dƒƒZG dd„ deƒZdS )	z
Producer-Consumer Proxy.
é    )Úimplementer)Ú
interfacesc                   @   sz   e Zd ZdZdZdZdZdZdZdZ	dZ
dd„ Zdd„ Zd	d
„ Zdd„ Zdd„ Zdd„ Zdd„ Zdd„ Zdefdd„ZdS )Ú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 ur|| _| | | j¡ d S 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| _ | jr| j ¡  d S d S ©NT)ÚpausedÚproducerÚpauseProducing©r
   r   r   r   r   +   s   ÿz)BasicProducerConsumerProxy.pauseProducingc                 C   sX   d| _ | jr| j d | j¡¡ g | jd d …< n| jsd| _| jd ur*| j ¡  d S d S )NFÚ T)	r   r   r   ÚwriteÚjoinr	   ÚoutstandingPullr   ÚresumeProducingr   r   r   r   r   0   s   
ÿz*BasicProducerConsumerProxy.resumeProducingc                 C   s*   | j d ur
| j  ¡  | jd ur| `d S d S r   )r   ÚstopProducingr   r   r   r   r   r   =   s
   


ÿz(BasicProducerConsumerProxy.stopProducingc                 C   sF   | j s	| js| js| j |¡ d S | jd ur!| j |¡ d| _d S d S ©NF)r   r	   r   r   Úappendr   r   ©r
   Údatar   r   r   r   E   s   

þz BasicProducerConsumerProxy.writec                 C   s    | j d ur
| j  ¡  |  ¡  d S r   )r   ÚfinishÚunregisterProducerr   r   r   r   r   N   s   

z!BasicProducerConsumerProxy.finishc                 C   s   || _ || _d S r   )r   ÚproducerIsStreaming©r
   r   Ú	streamingr   r   r   r   S   s   
z+BasicProducerConsumerProxy.registerProducerc                 C   s*   | j d ur	| ` | `| jr| j ¡  d S d S r   )r   r   r   r   r   r   r   r   r   W   s   
ÿz-BasicProducerConsumerProxy.unregisterProducerÚreturnc                 C   s"   d| j › dt| ƒd›d| j› d�S )Nú<ú@Úxz around ú>)Ú	__class__Úidr   r   r   r   r   Ú__repr__^   s   "z#BasicProducerConsumerProxy.__repr__)Ú__name__Ú
__module__Ú__qualname__Ú__doc__r   r   r   r	   r   r   Ústoppedr   r   r   r   r   r   r   r   Ústrr)   r   r   r   r   r      s$    	r   c                   @   sL   e Zd ZdZd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 )Ú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   o   s   
z$ProducerConsumerProxy.pauseProducingc                 C   sþ   d| _ | jr5d | j¡}|  |¡}|t|ƒk r-||d … }| jr$J dƒ‚|g| jd d …< n
g | jd d …< nd}| jrI|rI| jsI| jd urI| j ¡  | jsP| | _	| j
d ur{tdd„ | jD ƒƒ}| jrq|| jk rqd| _| j
 ¡  d S | j	r}| j
 ¡  d S d S d S )NFr   ú.Streaming producer did not write all its data.r   c                 s   ó   � | ]}t |ƒV  qd S r   ©Úlen©Ú.0Úsr   r   r   Ú	<genexpr>�   ó   € z8ProducerConsumerProxy.resumeProducing.<locals>.<genexpr>)r   r   r   Ú_writeSomeDatar4   r	   Úunregisteredr   r   r   r   ÚsumÚproducerPausedÚ
bufferSizer   )r
   r   Ú	bytesSentÚunsentÚbytesBufferedr   r   r   r   t   s@   
ÿþÿþý


ðz%ProducerConsumerProxy.resumeProducingc                 C   sÄ   | j s	| js| js| j |¡ n+| jd ur;| jrJ dƒ‚|  |¡}d| _|t|ƒks;| jr1J dƒ‚| j ||d … ¡ | jd ur\| j	r^t
dd„ | jD ƒƒ}|| jkr`| j ¡  d| _d S d S d S d S )Nz9Writing fresh data to consumer before my buffer is empty!Fr1   c                 s   r2   r   r3   r5   r   r   r   r8   ¸   r9   z.ProducerConsumerProxy.write.<locals>.<genexpr>T)r   r	   r   r   r   r   r:   r4   r   r   r<   r>   r   r=   )r
   r   r?   rA   r   r   r   r   ¡   s,   
ÿþ
ÿþ


ûzProducerConsumerProxy.writec                 C   s(   d| _ t | ||¡ |s| ¡  d S d S r   )r;   r   r   r   r    r   r   r   r   ¾   s
   ÿz&ProducerConsumerProxy.registerProducerc                 C   s:   | j d ur	| ` | `d| _| jr| js| j ¡  d S d S d S r   )r   r   r;   r   r   r   r   r   r   r   r   Ä   s   
ÿz(ProducerConsumerProxy.unregisterProducerc                 C   s"   | j du rdS | j  |¡ t|ƒS )z`Write as much of this data as possible.

        @returns: The number of bytes written.
        Nr   )r   r   r4   r   r   r   r   r:   Ì   s   
z$ProducerConsumerProxy._writeSomeDataN)r*   r+   r,   r-   r>   r=   r;   r   r   r   r   r   r:   r   r   r   r   r0   b   s    -r0   N)	r-   Úzope.interfacer   Útwisted.internetr   Ú	IProducerÚ	IConsumerr   r0   r   r   r   r   Ú<module>   s   S