
    \d                         d Z ddlmZ ddlmZ  eej        ej                   G d d                      Z G d de          ZdS )	z
Producer-Consumer Proxy.
    )implementer)
interfacesc                   j    e Zd ZdZdZdZdZdZdZdZ	dZ
d Zd Zd Zd Zd	 Zd
 Zd Zd Zdef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                 `    g | _         |$|| _        |                    | | j                   d S d S N)_bufferconsumerregisterProduceriAmStreaming)selfr
   s     5lib/python3.11/site-packages/twisted/protocols/pcp.py__init__z#BasicProducerConsumerProxy.__init__#   s=    $DM%%dD,=>>>>>      c                 X    d| _         | j        r| j                                         d S d S NT)pausedproducerpauseProducingr   s    r   r   z)BasicProducerConsumerProxy.pauseProducing+   s7    = 	+M((*****	+ 	+r   c                     d| _         | j        r?| j                            d                    | j                             g | j        d d <   n| j        sd| _        | j        | j                                         d S d S )NF T)	r   r	   r
   writejoinr   outstandingPullr   resumeProducingr   s    r   r   z*BasicProducerConsumerProxy.resumeProducing0   s    < 	,M 5 5666 DLOO$ ,'+$=$M))+++++ %$r   c                 \    | j         | j                                          | j        | `d S d S r   )r   stopProducingr
   r   s    r   r   z(BasicProducerConsumerProxy.stopProducing=   s7    =$M'')))=$ %$r   c                     | j         s| j        s#| j        s| j                            |           d S | j        #| j                            |           d| _        d S d S NF)r   r   r   r	   appendr
   r   r   datas     r   r   z BasicProducerConsumerProxy.writeE   sp    ; 	)t0 	)9M 	)L%%%%%]&M%%%#(D    '&r   c                 n    | j         | j                                          |                                  d S r   )r
   finishunregisterProducerr   s    r   r%   z!BasicProducerConsumerProxy.finishN   s6    =$M  """!!!!!r   c                 "    || _         || _        d S r   )r   producerIsStreamingr   r   	streamings      r   r   z+BasicProducerConsumerProxy.registerProducerS   s     #,   r   c                 `    | j         | ` | `| j        r| j                                         d S d S r   )r   r(   r
   r&   r   s    r   r&   z-BasicProducerConsumerProxy.unregisterProducerW   sC    =$(= 	/M,,.....	/ 	/r   returnc                 J    d| j          dt          |           dd| j         dS )N<@xz around >)	__class__idr
   r   s    r   __repr__z#BasicProducerConsumerProxy.__repr__^   s0    H4>HHBtHHHHHHHHHr   )__name__
__module____qualname____doc__r
   r   r(   r   r   r   stoppedr   r   r   r   r   r%   r   r&   strr4    r   r   r   r      s        	 	 HHLOFG? ? ?+ + +
, , ,  ) ) )" " "
- - -/ / /I# I I I I I Ir   r   c                   B    e Zd ZdZdZdZdZd Zd Zd Z	d Z
d Zd	 Zd
S )ProducerConsumerProxyzProducerConsumerProxy 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                     d| _         d S r   )r   r   s    r   r   z$ProducerConsumerProxy.pauseProducingo   s     r   c                    d| _         | j        rxd                    | j                  }|                     |          }|t	          |          k     r)||d          }| j        r
J d            |g| j        d d <   ng | j        d d <   nd}| j        r)|r'| j        s | j        | j                                         | j        s| | _	        | j
        tt          d | j        D                       }| j        r-|| j        k     r"d| _        | j
                                         d S | j	        r| j
                                         d S d S d S )NFr   .Streaming producer did not write all its data.r   c              3   4   K   | ]}t          |          V  d S r   len.0ss     r   	<genexpr>z8ProducerConsumerProxy.resumeProducing.<locals>.<genexpr>   (      ==1A======r   )r   r	   r   _writeSomeDatarC   r   unregisteredr
   r&   r   r   sumproducerPaused
bufferSizer   )r   r#   	bytesSentunsentbytesBuffereds        r   r   z%ProducerConsumerProxy.resumeProducingt   s   < 	774<((D++D11I3t99$$ijj))D DCD D D $*(QQQ"$QQQI 	/	/ L	/ )M,,...  	1'0=D =$=======M " 0(G(G&+#--/////% 0 --/////! %$0 0r   c                    | j         s| j        s"| j        s| j                            |           nz| j        s| j        r
J d            |                     |          }d| _        |t          |          k    s3| j        r
J d            | j                            ||d                     | j        R| j	        rMt          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!Fr@   c              3   4   K   | ]}t          |          V  d S r   rB   rD   s     r   rG   z.ProducerConsumerProxy.write.<locals>.<genexpr>   rH   r   T)r   r   r   r	   r!   r
   rI   rC   r   r(   rK   rM   r   rL   )r   r#   rN   rP   s       r   r   zProducerConsumerProxy.write   sN   ; 	6t0 	69M 	6L%%%%]&LK KJK K K ++D11I#(D D		)))D DCD D D ##D$4555M%4+C%=======M//,,...&*### &%%%//r   c                 |    d| _         t                              | ||           |s|                                 d S d S r    )rJ   r   r   r   r)   s      r   r   z&ProducerConsumerProxy.registerProducer   sL    !"33D(INNN 	'$$&&&&&	' 	'r   c                     | j         | ` | `d| _        | j        r"| j        s| j                                         d S d S d S r   )r   r(   rJ   r
   r	   r&   r   s    r   r&   z(ProducerConsumerProxy.unregisterProducer   s^    =$( = 	/ 	/M,,.....	/ 	/ 	/ 	/r   c                 f    | j         dS | j                             |           t          |          S )z`Write as much of this data as possible.

        @returns: The number of bytes written.
        Nr   )r
   r   rC   r"   s     r   rI   z$ProducerConsumerProxy._writeSomeData   s3    
 = 1D!!!4yyr   N)r5   r6   r7   r8   rM   rL   rJ   r   r   r   r   r&   rI   r;   r   r   r=   r=   b   s          "JNL  
+0 +0 +0Z+ + +:' ' '/ / /    r   r=   N)	r8   zope.interfacer   twisted.internetr   	IProducer	IConsumerr   r=   r;   r   r   <module>rZ      s   
  ' & & & & & ' ' ' ' ' ' Z!:#788PI PI PI PI PI PI PI 98PIfr r r r r6 r r r r rr   