class ConsumingFailed(Exception): passclass IConsumer(Interface): """Consume data.""" def write(data): """Return None if internal buffer still has room, otherwise return Deferred. Deferred will trigger when buffer has room again. It returns how much room is available in buffer, or fails with ConsumingFailed. """ # maybe writeSequence too. def bufferSize(): """Return how much room is available for writing."""class IByteConsumer(IConsumer): """Consumer of bytes."""class IProducer(Interface): def produce(consumer): """Start producing to consumer. Return Deferred of when producing is done, possibly with error indicating failure. """ # Maybe this too? Deferred cancellation would be good alternative. #def cancel(): # ""# In theory above is all we need, but some utility classes are# helpful:class IteratorProducer(object): """Produce an iterable.""" implements(IProducer) def __init__(self, iterable): self.iterable = iterable def publish(self, consumer): def gen(): for chunk in self.iterable: yield self.consumer.write(chunk) return coiterate(gen())class FileProducer(object): """Produce a file.""" implements(IProducer) def __init__(self, filename): self.filename = filename def publish(self, consumer): try: f = file(filename, "rb") except: return defer.fail() def gen(): while True: s = consumer.bufferSize() if s == 0: yield d = f.read(s) if not d: return yield self.consumer.write(d) return coiterate(gen())# Example streaming producerclass ProducingProtocol(Protocol): """Pushes data it receives to consumer.""" def __init__(self, consumer): self.consumer = consumer def dataReceived(self, data): result = self.consumer.write(data) if not result: return # we got deferred, so stop producing: self.stopReading() result.addCallbacks(lambda x: self.startReading(), self._cbConsumerFailed) def _cbConsumerFailed(self, reason): log.err(reason) self.transport.loseConnection()# other potential uses: limited-buffer consumer adapter that throws# exception if too much memory is being buffered. consumers can now# have more sophisticated policies than simply "I'm empty/I'm full"# notification we currently have.