ó [³XMc@sSdZddlmZddlmZdd d„ƒYZdefd„ƒYZdS( s Producer-Consumer Proxy. iÿÿÿÿ(t implements(t interfacestBasicProducerConsumerProxycBsœeZdZeejejƒd Zd Z d Z e Z e Ze Ze Zd„Zd„Zd„Zd„Zd„Zd„Zd„Zd„Zd „ZRS( sa 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} @ivar consumer: the Consumer I publish to. @type consumer: L{IConsumer} @ivar paused: As a Producer, am I paused? @type paused: bool cCs8g|_|dk r4||_|j||jƒndS(N(t_buffertNonetconsumertregisterProducert iAmStreaming(tselfR((s9/usr/lib/python2.7/dist-packages/twisted/protocols/pcp.pyt__init__#s   cCs&t|_|jr"|jjƒndS(N(tTruetpausedtproducertpauseProducing(R((s9/usr/lib/python2.7/dist-packages/twisted/protocols/pcp.pyR +s  cCsst|_|jr;|jjdj|jƒƒg|j(n|jsPt|_n|j dk ro|j j ƒndS(Nt( tFalseR RRtwritetjoinRR toutstandingPullR RtresumeProducing(R((s9/usr/lib/python2.7/dist-packages/twisted/protocols/pcp.pyR0s     cCs;|jdk r|jjƒn|jdk r7|`ndS(N(R Rt stopProducingR(R((s9/usr/lib/python2.7/dist-packages/twisted/protocols/pcp.pyR=scCs_|js|j r0|j r0|jj|ƒn+|jdk r[|jj|ƒt|_ndS(N( R RRRtappendRRRR(Rtdata((s9/usr/lib/python2.7/dist-packages/twisted/protocols/pcp.pyREs cCs-|jdk r|jjƒn|jƒdS(N(RRtfinishtunregisterProducer(R((s9/usr/lib/python2.7/dist-packages/twisted/protocols/pcp.pyRNscCs||_||_dS(N(R tproducerIsStreaming(RR t streaming((s9/usr/lib/python2.7/dist-packages/twisted/protocols/pcp.pyRSs cCs;|jdk r|`|`n|jr7|jjƒndS(N(R RRRR(R((s9/usr/lib/python2.7/dist-packages/twisted/protocols/pcp.pyRWs   cCsd|jt|ƒ|jfS(Ns<%s@%x around %s>(t __class__tidR(R((s9/usr/lib/python2.7/dist-packages/twisted/protocols/pcp.pyt__repr__^sN(t__name__t __module__t__doc__RRt IProducert IConsumerRRR RR RRRR tstoppedR R RRRRRRR(((s9/usr/lib/python2.7/dist-packages/twisted/protocols/pcp.pyRs$       tProducerConsumerProxycBs^eZdZdddZeZeZd„Zd„Zd„Z d„Z d„Z d„Z RS( s˜ProducerConsumerProxy with a finite buffer. When my buffer fills up, I have my parent Producer pause until my buffer has room in it again. icCs t|_dS(N(R R (R((s9/usr/lib/python2.7/dist-packages/twisted/protocols/pcp.pyR nscCs[t|_|jrdj|jƒ}|j|ƒ}|t|ƒkrr||}|j sbtd‚|g|j(q…g|j(nd}|jr½|r½|j r½|j dk r½|j j ƒn|jsÓ| |_ n|j dk rWtg|jD]}t|ƒ^qïƒ}|jr;||jkr;t|_|j jƒqW|j rW|j jƒqWndS(NRs.Streaming producer did not write all its data.i(RR RRt_writeSomeDatatlenRtAssertionErrort unregisteredRRRRR tsumtproducerPausedt bufferSizeR(RRt bytesSenttunsenttst bytesBuffered((s9/usr/lib/python2.7/dist-packages/twisted/protocols/pcp.pyRss.       (  cCs|js|j r0|j r0|jj|ƒny|jdk r©|j sRtd‚|j|ƒ}t |_|t |ƒks©|j std‚|jj||ƒq©n|j dk r|j rt g|jD]}t |ƒ^q΃}||jkr|j jƒt|_qndS(Ns9Writing fresh data to consumer before my buffer is empty!s.Streaming producer did not write all its data.(R RRRRRRR'R%RR&R RR)R+R R R*(RRR,R.R/((s9/usr/lib/python2.7/dist-packages/twisted/protocols/pcp.pyR›s    ( cCs3t|_tj|||ƒ|s/|jƒndS(N(RR(RRR(RR R((s9/usr/lib/python2.7/dist-packages/twisted/protocols/pcp.pyR¶s cCsN|jdk r|`|`nt|_|jrJ|j rJ|jjƒndS(N(R RRR R(RRR(R((s9/usr/lib/python2.7/dist-packages/twisted/protocols/pcp.pyR¼s   cCs-|jdkrdS|jj|ƒt|ƒS(s`Write as much of this data as possible. @returns: The number of bytes written. iN(RRRR&(RR((s9/usr/lib/python2.7/dist-packages/twisted/protocols/pcp.pyR%Äsi( RRR R+RR*R(R RRRRR%(((s9/usr/lib/python2.7/dist-packages/twisted/protocols/pcp.pyR$bs  (   N((R tzope.interfaceRttwisted.internetRRR$(((s9/usr/lib/python2.7/dist-packages/twisted/protocols/pcp.pytsT