U
    Â“«j  ã                   @   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edœ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 k	r"|| _| | | j¡ d S ©N)Ú_bufferÚconsumerÚregisterProducerÚiAmStreaming)Úselfr   © r   ú9/tmp/pip-unpacked-wheel-59kc99xd/twisted/protocols/pcp.pyÚ__init__#   s    z#BasicProducerConsumerProxy.__init__c                 C   s   d| _ | jr| j ¡  d S ©NT)ÚpausedÚproducerÚpauseProducing©r
   r   r   r   r   +   s    z)BasicProducerConsumerProxy.pauseProducingc                 C   sT   d| _ | jr0| j d | j¡¡ g | jd d …< n| js<d| _| jd k	rP| j ¡  d S )NFÚ T)	r   r   r   ÚwriteÚjoinr	   ÚoutstandingPullr   ÚresumeProducingr   r   r   r   r   0   s    
z*BasicProducerConsumerProxy.resumeProducingc                 C   s&   | j d k	r| j  ¡  | jd k	r"| `d S r   )r   ÚstopProducingr   r   r   r   r   r   =   s    


z(BasicProducerConsumerProxy.stopProducingc                 C   s@   | j s| js | js | j |¡ n| jd k	r<| j |¡ d| _d S ©NF)r   r	   r   r   Úappendr   r   ©r
   Údatar   r   r   r   E   s
    
z BasicProducerConsumerProxy.writec                 C   s    | j d k	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   Z	streamingr   r   r   r   S   s    z+BasicProducerConsumerProxy.registerProducerc                 C   s&   | j d k	r| ` | `| jr"| j ¡  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rjd | j¡}|  |¡}|t|ƒk rZ||d … }| jrHtdƒ‚|g| jd d …< qng | jd d …< nd}| jr’|r’| js’| jd k	r’| j 	¡  | js | | _
| jd k	rðtdd„ | jD ƒƒ}| jrà|| jk ràd| _| j ¡  n| j
rð| j ¡  d S )NFr   ú.Streaming producer did not write all its data.r   c                 s   s   | ]}t |ƒV  qd S r   ©Úlen©Ú.0Úsr   r   r   Ú	<genexpr>�   s     z8ProducerConsumerProxy.resumeProducing.<locals>.<genexpr>)r   r   r   Ú_writeSomeDatar2   r	   ÚAssertionErrorÚunregisteredr   r   r   r   ÚsumÚproducerPausedÚ
bufferSizer   )r
   r   Ú	bytesSentZunsentÚbytesBufferedr   r   r   r   t   s>    
ÿþÿþýü

z%ProducerConsumerProxy.resumeProducingc                 C   s¸   | j s| js | js | j |¡ nV| jd k	rv| jr8tdƒ‚|  |¡}d| _|t|ƒksv| jrbtdƒ‚| j ||d … ¡ | j	d k	r´| j
r´tdd„ | jD ƒƒ}|| jkr´| j	 ¡  d| _d S )Nz9Writing fresh data to consumer before my buffer is empty!Fr0   c                 s   s   | ]}t |ƒV  qd S r   r1   r3   r   r   r   r6   ¸   s     z.ProducerConsumerProxy.write.<locals>.<genexpr>T)r   r	   r   r   r   r   r8   r7   r2   r   r   r:   r<   r   r;   )r
   r   r=   r>   r   r   r   r   ¡   s(    
ÿþ
ÿþ

zProducerConsumerProxy.writec                 C   s$   d| _ t | ||¡ |s | ¡  d S r   )r9   r   r   r   r    r   r   r   r   ½   s    z&ProducerConsumerProxy.registerProducerc                 C   s2   | j d k	r| ` | `d| _| jr.| js.| j ¡  d S r   )r   r   r9   r   r   r   r   r   r   r   r   Ã   s    
z(ProducerConsumerProxy.unregisterProducerc                 C   s"   | j dkrdS | j  |¡ t|ƒS )z`Write as much of this data as possible.

        @returns: The number of bytes written.
        Nr   )r   r   r2   r   r   r   r   r7   Ë   s    
z$ProducerConsumerProxy._writeSomeDataN)r)   r*   r+   r,   r<   r;   r9   r   r   r   r   r   r7   r   r   r   r   r/   b   s   -r/   N)	r,   Zzope.interfacer   Ztwisted.internetr   Z	IProducerZ	IConsumerr   r/   r   r   r   r   Ú<module>   s
   S