doc/core/examples/streaming.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) Twisted Matrix Laboratories.
   4 # See LICENSE for details.
   5
   6 """
   7 This is a sample implementation of a Twisted push producer/consumer system. It
   8 consists of a TCP server which asks the user how many random integers they
   9 want, and it sends the result set back to the user, one result per line,
  10 and finally closes the connection.
  11 """
  12
  13 from sys import stdout
  14 from random import randrange
  15
  16 from zope.interface import implements
  17 from twisted.python.log import startLogging
  18 from twisted.internet import interfaces, reactor
  19 from twisted.internet.protocol import Factory
  20 from twisted.protocols.basic import LineReceiver
  21
  22
  23 class Producer(object):
  24     """
  25     Send back the requested number of random integers to the client.
  26     """
  27
  28     implements(interfaces.IPushProducer)
  29
  30     def __init__(self, proto, count):
  31         self._proto = proto
  32         self._goal = count
  33         self._produced = 0
  34         self._paused = False
  35
  36     def pauseProducing(self):
  37         """
  38         When we've produced data too fast, pauseProducing() will be called
  39         (reentrantly from within resumeProducing's sendLine() method, most
  40         likely), so set a flag that causes production to pause temporarily.
  41         """
  42         self._paused = True
  43         print 'Pausing connection from %s' % self._proto.transport.getPeer()
  44
  45     def resumeProducing(self):
  46         """
  47         Resume producing integers.
  48
  49         This tells the push producer to (re-)add itself to the main loop and
  50         produce integers for its consumer until the requested number of integers
  51         were returned to the client.
  52         """
  53         self._paused = False
  54
  55         while not self._paused and self._produced < self._goal:
  56             next_int = randrange(0, 10000)
  57             self._proto.sendLine('%d' % next_int)
  58             self._produced += 1
  59
  60         if self._produced == self._goal:
  61             self._proto.transport.unregisterProducer()
  62             self._proto.transport.loseConnection()
  63
  64     def stopProducing(self):
  65         """
  66         When a consumer has died, stop producing data for good.
  67         """
  68         self._produced = self._goal
  69
  70
  71 class ServeRandom(LineReceiver):
  72     """
  73     Serve up random integers.
  74     """
  75
  76     def connectionMade(self):
  77         """
  78         Once the connection is made we ask the client how many random integers
  79         the producer should return.
  80         """
  81         print 'Connection made from %s' % self.transport.getPeer()
  82         self.sendLine('How many random integers do you want?')
  83
  84     def lineReceived(self, line):
  85         """
  86         This checks how many random integers the client expects in return and
  87         tells the producer to start generating the data.
  88         """
  89         count = int(line.strip())
  90         print 'Client requested %d random integers!' % count
  91         producer = Producer(self, count)
  92         self.transport.registerProducer(producer, True)
  93         producer.resumeProducing()
  94
  95     def connectionLost(self, reason):
  96         print 'Connection lost from %s' % self.transport.getPeer()
  97
  98
  99 startLogging(stdout)
 100 factory = Factory()
 101 factory.protocol = ServeRandom
 102 reactor.listenTCP(1234, factory)
 103 reactor.run()