<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.0 Transitional//EN">
<HTML><HEAD><TITLE>Learning about IPushProducer</TITLE>
<META http-equiv=Content-Type content="text/html; charset=us-ascii">
<META content="MSHTML 6.00.2900.3059" name=GENERATOR></HEAD>
<BODY>
<DIV dir=ltr align=left><SPAN class=483154119-12032007><FONT face="Courier New" 
color=#0000ff size=2>Hi.</FONT></SPAN></DIV>
<DIV dir=ltr align=left><SPAN class=483154119-12032007><FONT face="Courier New" 
color=#0000ff size=2></FONT></SPAN>&nbsp;</DIV>
<DIV dir=ltr align=left><SPAN class=483154119-12032007><FONT face="Courier New" 
color=#0000ff size=2>Anyone have any pointers as to how I can get some of my 
questions answered below?&nbsp; I had hoped to get some response.&nbsp; Did I 
not use the proper etiquitte?&nbsp; Or there is some expert on the IPushProducer 
mechanism or the author of page </FONT></SPAN></DIV>
<DIV dir=ltr align=left><SPAN class=483154119-12032007><FONT face="Courier New" 
color=#0000ff size=2></FONT></SPAN>&nbsp;</DIV>
<DIV dir=ltr align=left><SPAN class=483154119-12032007><FONT face="Courier New" 
color=#0000ff size=2><A 
href="http://twistedmatrix.com/projects/core/documentation/howto/producers.html">http://twistedmatrix.com/projects/core/documentation/howto/producers.html</A></FONT></SPAN></DIV>
<DIV><FONT face="Courier New" color=#0000ff size=2></FONT>&nbsp;</DIV>
<DIV><FONT color=#606420></FONT><SPAN class=483154119-12032007></SPAN><FONT 
face="Courier New"><FONT color=#0000ff><FONT size=2>t<SPAN 
class=483154119-12032007>hat I can be referred to that perhaps isn't reading 
this list?</SPAN></FONT></FONT></FONT><BR></DIV>
<DIV><SPAN class=483154119-12032007><FONT face="Courier New" color=#0000ff 
size=2>Thanks,</FONT></SPAN></DIV>
<DIV><SPAN class=483154119-12032007><FONT face="Courier New" color=#0000ff 
size=2>Benjamin Rutt</FONT></SPAN></DIV>
<BLOCKQUOTE dir=ltr style="MARGIN-RIGHT: 0px">
  <DIV class=OutlookMessageHeader lang=en-us dir=ltr align=left>
  <HR tabIndex=-1>
  <FONT face=Tahoma size=2><B>From:</B> twisted-python-bounces@twistedmatrix.com 
  [mailto:twisted-python-bounces@twistedmatrix.com] <B>On Behalf Of </B>Rutt, 
  Benjamin<BR><B>Sent:</B> Tuesday, March 06, 2007 12:04 PM<BR><B>To:</B> 
  twisted-python@twistedmatrix.com<BR><B>Subject:</B> [Twisted-Python] Learning 
  about IPushProducer<BR></FONT><BR></DIV>
  <DIV></DIV><!-- Converted from text/rtf format -->
  <P><FONT face="Courier New" size=2>When running the following code (my 2nd 
  twisted program!), it works as I had hoped - it doesn't starve any clients 
  that want to receive data back, even with a simultaneously active really long 
  streaming server-to-client communication (i.e. one piggy client asking for 
  millions of bytes).&nbsp; i.e. another client can get in and ask for just a 
  few bytes while a large payload is being delivered to a different 
  client.&nbsp; Which is great!</FONT></P>
  <P><FONT face="Courier New" size=2>Here's a sample interaction from the client 
  side:</FONT> </P>
  <P><FONT face="Courier New" size=2>$ telnet localhost 8007</FONT> <BR><FONT 
  face="Courier New" size=2>Trying 127.0.0.1...</FONT> <BR><FONT 
  face="Courier New" size=2>Connected to localhost.</FONT> <BR><FONT 
  face="Courier New" size=2>Escape character is '^]'.</FONT> <BR><FONT 
  face="Courier New" size=2>1</FONT> <BR><FONT face="Courier New" 
  size=2>x</FONT> <BR><FONT face="Courier New" size=2>2</FONT> <BR><FONT 
  face="Courier New" size=2>xx</FONT> <BR><FONT face="Courier New" 
  size=2>3</FONT> <BR><FONT face="Courier New" size=2>xxx</FONT> <BR><FONT 
  face="Courier New" size=2>10</FONT> <BR><FONT face="Courier New" 
  size=2>xxxxxxxxxx</FONT> <BR><FONT face="Courier New" size=2>99999</FONT> 
  <BR><FONT face="Courier New" 
  size=2>xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx</FONT></P>
  <P><FONT face="Courier New" size=2>[...lots of x's...]</FONT> <BR><FONT 
  face="Courier New" size=2>xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx</FONT> <BR><FONT 
  face="Courier New" size=2>bye</FONT> <BR><FONT face="Courier New" 
  size=2>Connection closed by foreign host.</FONT> <BR><FONT face="Courier New" 
  size=2>$ </FONT></P>
  <P><FONT face="Courier New" size=2>So I have 2 questions on my code:</FONT> 
  </P>
  <P><FONT face="Courier New" size=2>1) am I doing anything wrong in setting up 
  the plumbing?</FONT> <BR><FONT face="Courier New" size=2>2) does 
  pauseProducing() get called by another thread whilst resumeProducing() is 
  running?&nbsp; (I believe it must, otherwise my resumeProducing() would only 
  be entered once).&nbsp; If so I should have an appropriate mutex around the 
  read/write of self.pause, no?</FONT></P>
  <P><FONT face="Courier New" size=2>Here is the code, and output from the 
  server is at the end.&nbsp; Thanks -- Benjamin</FONT> </P>
  <P><FONT face="Courier New" size=2>#!/usr/bin/env python</FONT> <BR><FONT 
  face="Courier New" size=2>import os, os.path, sys, re, commands, pickle, 
  tempfile, getopt, datetime</FONT> <BR><FONT face="Courier New" size=2>import 
  socket, string, random, time, traceback, shutil, popen2</FONT> </P>
  <P><FONT face="Courier New" size=2>from zope.interface import 
  implements</FONT> <BR><FONT face="Courier New" size=2>from twisted.internet 
  import protocol, defer, interfaces, error, reactor</FONT> <BR><FONT 
  face="Courier New" size=2>from twisted.internet.protocol import Protocol, 
  Factory</FONT> <BR><FONT face="Courier New" size=2>from 
  twisted.protocols.basic import LineReceiver</FONT> </P>
  <P><FONT face="Courier New" size=2>class NonStarvingXGiver:</FONT> <BR><FONT 
  face="Courier New" size=2>&nbsp;&nbsp;&nbsp; 
  implements(interfaces.IPushProducer)</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp; def __init__(self, howmany, consumer):</FONT> 
  <BR><FONT face="Courier New" size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
  self.howmany = howmany</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; self.sent_already = 0</FONT> 
  <BR><FONT face="Courier New" size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
  self.paused = False</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; self.consumer = 
  consumer</FONT> <BR><FONT face="Courier New" size=2>&nbsp;&nbsp;&nbsp; def 
  beginSendingXs(self):</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; self.deferred = deferred = 
  defer.Deferred()</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
  self.consumer.registerProducer(self, False)</FONT> <BR><FONT 
  face="Courier New" size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; return 
  deferred</FONT> <BR><FONT face="Courier New" size=2>&nbsp;&nbsp;&nbsp; def 
  pauseProducing(self):</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; print 'pauseProducing: 
  invoked'</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; self.paused = True</FONT> 
  <BR><FONT face="Courier New" size=2>&nbsp;&nbsp;&nbsp; def 
  resumeProducing(self):</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; print 'resumeProducing: 
  invoked'</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; self.paused = False</FONT> 
  <BR><FONT face="Courier New" size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
  maxchunksz = 1024</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; while not self.paused and 
  self.howmany &gt; self.sent_already:</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
  chunksz = min(maxchunksz, self.howmany - self.sent_already)</FONT> <BR><FONT 
  face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
  self.consumer.write('x' * chunksz)</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
  self.sent_already += chunksz</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if self.howmany == 
  self.sent_already:</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
  self.consumer.write('\n')</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
  self.consumer.unregisterProducer()</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
  print 'resumeProducing: exiting for the last time'</FONT> <BR><FONT 
  face="Courier New" size=2>&nbsp;&nbsp;&nbsp; def stopProducing(self):</FONT> 
  <BR><FONT face="Courier New" size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
  print 'stopProducing: invoked'</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
  self.consumer.unregisterProducer()</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; </FONT><BR><FONT 
  face="Courier New" size=2>class xgiver(LineReceiver):</FONT> <BR><FONT 
  face="Courier New" size=2>&nbsp;&nbsp;&nbsp; def lineReceived(self, 
  howmany):</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; print 'got line [%s] from 
  client [%s]' % (howmany,</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
  self.transport.getPeer())</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if howmany == 'bye':</FONT> 
  <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
  print 'goodbye to', self.transport.getPeer()</FONT> <BR><FONT 
  face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
  self.transport.loseConnection()</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
  return</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; try:</FONT> <BR><FONT 
  face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
  howmany = int(howmany)</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; s = 
  NonStarvingXGiver(howmany, self.transport)</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
  s.beginSendingXs()</FONT> <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; except Exception, ex:</FONT> 
  <BR><FONT face="Courier New" 
  size=2>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
  self.transport.write("invalid input " + howmany + "\n")</FONT> </P>
  <P><FONT face="Courier New" size=2># Next lines are magic:</FONT> <BR><FONT 
  face="Courier New" size=2>factory = Factory()</FONT> <BR><FONT 
  face="Courier New" size=2>factory.protocol = xgiver</FONT> </P>
  <P><FONT face="Courier New" size=2># 8007 is the port you want to run under. 
  Choose something &gt;1024</FONT> <BR><FONT face="Courier New" 
  size=2>reactor.listenTCP(8007, factory)</FONT> <BR><FONT face="Courier New" 
  size=2>reactor.run()</FONT> </P><BR>
  <P><FONT face="Courier New" 
  size=2>-------------------------------------------------------------------</FONT> 
  <BR><FONT face="Courier New" size=2>Server output:</FONT> </P>
  <P><FONT face="Courier New" size=2>$ ./xgiver.py </FONT><BR><FONT 
  face="Courier New" size=2>got line [1] from client [IPv4Address(TCP, 
  '127.0.0.1', 51007)]</FONT> <BR><FONT face="Courier New" 
  size=2>resumeProducing: invoked</FONT> <BR><FONT face="Courier New" 
  size=2>resumeProducing: exiting for the last time</FONT> <BR><FONT 
  face="Courier New" size=2>got line [2] from client [IPv4Address(TCP, 
  '127.0.0.1', 51007)]</FONT> <BR><FONT face="Courier New" 
  size=2>resumeProducing: invoked</FONT> <BR><FONT face="Courier New" 
  size=2>resumeProducing: exiting for the last time</FONT> <BR><FONT 
  face="Courier New" size=2>got line [3] from client [IPv4Address(TCP, 
  '127.0.0.1', 51007)]</FONT> <BR><FONT face="Courier New" 
  size=2>resumeProducing: invoked</FONT> <BR><FONT face="Courier New" 
  size=2>resumeProducing: exiting for the last time</FONT> <BR><FONT 
  face="Courier New" size=2>got line [10] from client [IPv4Address(TCP, 
  '127.0.0.1', 51007)]</FONT> <BR><FONT face="Courier New" 
  size=2>resumeProducing: invoked</FONT> <BR><FONT face="Courier New" 
  size=2>resumeProducing: exiting for the last time</FONT> <BR><FONT 
  face="Courier New" size=2>got line [99999] from client [IPv4Address(TCP, 
  '127.0.0.1', 51007)]</FONT> <BR><FONT face="Courier New" 
  size=2>resumeProducing: invoked</FONT> <BR><FONT face="Courier New" 
  size=2>pauseProducing: invoked</FONT> <BR><FONT face="Courier New" 
  size=2>resumeProducing: invoked</FONT> <BR><FONT face="Courier New" 
  size=2>resumeProducing: exiting for the last time</FONT> <BR><FONT 
  face="Courier New" size=2>got line [bye] from client [IPv4Address(TCP, 
  '127.0.0.1', 51007)]</FONT> <BR><FONT face="Courier New" size=2>goodbye to 
  IPv4Address(TCP, '127.0.0.1', 51007)</FONT> </P></BLOCKQUOTE></BODY></HTML>