Repository: twisted/tubes

Issue:
replace conversion of flow-function → factory with conversion of endpoint → fount of founts
Presently on trunk, tubes are hooked up to the outside world (i.e.: Twisted) by converting a "flow function", a function taking a fount and drain, to a Factory, which is then hooked up to Twisted.

However, this is limiting in one major regard: the backpressure of *not accepting future connections* is impossible to model properly.

A better way to conceptualize a listening socket is a fount whose `outputType` is an object with both `Fount` and `Drain` attributes.  This way, we can flow the stream of incoming connections to a "listening" drain, which can exert backpressure (most obviously by simply limiting the number of concurrently active connections).

As it happens, this is also a feature missing from Twisted which makes it hard to implement fairness in queueing.  By making tubes treat listening sockets and connected sockets consistently, we can open up a whole new area of correct-by-default behavior under high levels of load.

Discussion:

Patch:
diff --git a/docs/listings/echoflow.py b/docs/listings/echoflow.py
index beea5b3..d2f3c1a 100644
--- a/docs/listings/echoflow.py
+++ b/docs/listings/echoflow.py
@@ -1,15 +1,18 @@
-from tubes.protocol import factoryFromFlow
+from tubes.protocol import flowFountFromEndpoint
+from tubes.listening import Listener
 
 from twisted.internet.endpoints import serverFromString
-from twisted.internet.defer import Deferred
+from twisted.internet.defer import Deferred, inlineCallbacks
 
-def echoFlow(fount, drain):
-    fount.flowTo(drain)
+def echoFlow(flow):
+    flow.fount.flowTo(flow.drain)
 
+@inlineCallbacks
 def main(reactor, listenOn="stdio:"):
     endpoint = serverFromString(reactor, listenOn)
-    endpoint.listen(factoryFromFlow(echoFlow))
-    return Deferred()
+    flowFount = yield flowFountFromEndpoint(endpoint)
+    flowFount.flowTo(Listener(echoFlow))
+    yield Deferred()
 
 if __name__ == '__main__':
     from twisted.internet.task import react
diff --git a/docs/listings/echonetstrings.py b/docs/listings/echonetstrings.py
index bf05626..1287f0f 100644
--- a/docs/listings/echonetstrings.py
+++ b/docs/listings/echonetstrings.py
@@ -1,17 +1,21 @@
 from tubes.tube import Tube
 from tubes.framing import stringsToNetstrings
-from tubes.protocol import factoryFromFlow
+from tubes.protocol import flowFountFromEndpoint
+from tubes.listening import Listener
+
 from twisted.internet.endpoints import TCP4ServerEndpoint
-from twisted.internet.defer import Deferred
+from twisted.internet.defer import inlineCallbacks, Deferred()
 
-def echoTubeFactory(fount, drain):
-    return (fount.flowTo(Tube(stringsToNetstrings()))
-                 .flowTo(drain))
+def echoTubeFactory(flow):
+    return (flow.fount.flowTo(Tube(stringsToNetstrings()))
+                      .flowTo(flow.drain))
 
+@inlineCallbacks
 def main(reactor):
     endpoint = TCP4ServerEndpoint(reactor, 4321)
-    endpoint.listen(factoryFromFlow(echoTubeFactory))
-    return Deferred()
+    flowFount = yield flowFountFromEndpoint(endpoint)
+    flowFount.flowTo(Listener(echoTubeFactory))
+    yield Deferred()
 
 if __name__ == '__main__':
     from twisted.internet.task import react
diff --git a/docs/listings/portforward.py b/docs/listings/portforward.py
index ec723dc..0941743 100644
--- a/docs/listings/portforward.py
+++ b/docs/listings/portforward.py
@@ -1,21 +1,22 @@
-import os
+from tubes.protocol import flowFountFromEndpoint, flowFromEndpoint
+from tubes.listening import Listener
 
-from tubes.protocol import factoryFromFlow
 from twisted.internet.endpoints import serverFromString, clientFromString
-from twisted.internet.defer import Deferred
+from twisted.internet.defer import Deferred, inlineCallbacks
 
+@inlineCallbacks
 def main(reactor, listen="tcp:4321", connect="tcp:localhost:6543"):
     clientEndpoint = clientFromString(reactor, connect)
     serverEndpoint = serverFromString(reactor, listen)
 
-    def incomingTubeFactory(listeningFount, listeningDrain):
-        def outgoingTubeFactory(connectingFount, connectingDrain):
-            listeningFount.flowTo(connectingDrain)
-            connectingFount.flowTo(listeningDrain)
-        clientEndpoint.connect(factoryFromFlow(outgoingTubeFactory))
-
-    serverEndpoint.listen(factoryFromFlow(incomingTubeFactory))
-    return Deferred()
+    def incoming(listening):
+        def outgoing(connecting):
+            listening.fount.flowTo(connecting.drain)
+            connecting.fount.flowTo(listening.drain)
+        flowFromEndpoint(clientEndpoint).addCallback(outgoing)
+    flowFount = yield flowFountFromEndpoint(serverEndpoint)
+    flowFount.flowTo(Listener(incoming))
+    yield Deferred()
 
 if __name__ == '__main__':
     from twisted.internet.task import react
diff --git a/docs/listings/reversetube.py b/docs/listings/reversetube.py
index cf99b0c..4e9096b 100644
--- a/docs/listings/reversetube.py
+++ b/docs/listings/reversetube.py
@@ -1,6 +1,8 @@
-from tubes.protocol import factoryFromFlow
 from twisted.internet.endpoints import serverFromString
-from twisted.internet.defer import Deferred
+from twisted.internet.defer import Deferred, inlineCallbacks
+
+from tubes.protocol import flowFountFromEndpoint
+from tubes.listening import Listener
 from tubes.tube import tube, series
 
 @tube
@@ -8,15 +10,17 @@ class Reverser(object):
     def received(self, item):
         yield b"".join(reversed(item))
 
-def reverseFlow(fount, drain):
+def reverseFlow(flow):
     from tubes.framing import bytesToLines, linesToBytes
     lineReverser = series(bytesToLines(), Reverser(), linesToBytes())
-    fount.flowTo(lineReverser).flowTo(drain)
+    flow.fount.flowTo(lineReverser).flowTo(flow.drain)
 
+@inlineCallbacks
 def main(reactor, listenOn="stdio:"):
     endpoint = serverFromString(reactor, listenOn)
-    endpoint.listen(factoryFromFlow(reverseFlow))
-    return Deferred()
+    flowFount = yield flowFountFromEndpoint(endpoint)
+    flowFount.flowTo(Listener(reverseFlow))
+    yield Deferred()
 
 if __name__ == '__main__':
     from twisted.internet.task import react
diff --git a/docs/listings/rpn.py b/docs/listings/rpn.py
index fe6f96c..697957e 100644
--- a/docs/listings/rpn.py
+++ b/docs/listings/rpn.py
@@ -1,9 +1,10 @@
-from tubes.protocol import factoryFromFlow
 from tubes.itube import IFrame, ISegment
 from tubes.tube import tube, receiver
+from tubes.listening import Listener
 
 from twisted.internet.endpoints import serverFromString
-from twisted.internet.defer import Deferred
+from twisted.internet.defer import Deferred, inlineCallbacks
+from tubes.protocol import flowFountFromEndpoint
 
 class Calculator(object):
     def __init__(self):
@@ -79,15 +80,17 @@ def calculatorSeries():
         linesToBytes()
     )
 
-def mathFlow(fount, drain):
+def mathFlow(flow):
     processor = calculatorSeries()
-    nextDrain = fount.flowTo(processor)
-    nextDrain.flowTo(drain)
+    nextDrain = flow.fount.flowTo(processor)
+    nextDrain.flowTo(flow.drain)
 
+@inlineCallbacks
 def main(reactor, port="stdio:"):
     endpoint = serverFromString(reactor, port)
-    endpoint.listen(factoryFromFlow(mathFlow))
-    return Deferred()
+    flowFount = yield flowFountFromEndpoint(endpoint)
+    flowFount.flowTo(Listener(mathFlow))
+    yield Deferred()
 
 if __name__ == '__main__':
     from twisted.internet.task import react
diff --git a/sketches/amptube.py b/sketches/amptube.py
index 128d183..0a7e8e9 100644
--- a/sketches/amptube.py
+++ b/sketches/amptube.py
@@ -1,18 +1,19 @@
-
-
 from zope.interface import implementer
 
 from ampserver import Math
 
-from twisted.tubes.protocol import factoryFromFlow
 from twisted.internet.endpoints import serverFromString
 
+from twisted.internet.defer import Deferred, inlineCallbacks
 from twisted.internet import reactor
 
 from twisted.protocols.amp import AmpBox, IBoxSender
-from twisted.tubes.itube import ISegment
-from twisted.tubes.tube import Pump, series
-from twisted.tubes.framing import packedPrefixToStrings
+
+from tubes.protocol import flowFountFromEndpoint
+from tubes.listening import Listener
+from tubes.itube import ISegment
+from tubes.tube import Pump, series
+from tubes.framing import packedPrefixToStrings
 
 class StringsToBoxes(Pump):
 
@@ -96,14 +97,22 @@ class BoxConsumer(Pump):
 
 
 
-def mathFlow(fount, drain):
+def mathFlow(fount):
     fount.flowTo(series(packedPrefixToStrings(16), StringsToBoxes(),
-                        BoxConsumer(Math()), BoxesToData(), drain))
+                        BoxConsumer(Math()), BoxesToData(), fount.drain))
 
 
 
-serverEndpoint = serverFromString(reactor, "tcp:1234")
-serverEndpoint.listen(factoryFromFlow(mathFlow))
-from twisted.internet import reactor
-reactor.run()
+
+@inlineCallbacks
+def main():
+    serverEndpoint = serverFromString(reactor, "tcp:1234")
+    flowFount = yield flowFountFromEndpoint(serverEndpoint)
+    flowFount.flowTo(Listener(mathFlow))
+    yield Deferred()
+
+
+from twisted.interne.task import react
+from sys import argv
+react(main, argv[1:])
 
diff --git a/sketches/fanchat.py b/sketches/fanchat.py
index c26a013..92cdcb3 100644
--- a/sketches/fanchat.py
+++ b/sketches/fanchat.py
@@ -5,14 +5,15 @@ from json import loads, dumps
 from zope.interface.common import IMapping
 
 from twisted.internet.endpoints import serverFromString
-from twisted.internet.defer import Deferred
+from twisted.internet.defer import Deferred, inlineCallbacks
 
-from twisted.tubes.routing import Router, Routed, to
-from twisted.tubes.protocol import factoryFromFlow
-from twisted.tubes.itube import IFrame
-from twisted.tubes.tube import series, tube, receiver
-from twisted.tubes.framing import bytesToLines, linesToBytes
-from twisted.tubes.fan import Out, In
+from tubes.routing import Router, Routed, to
+from tubes.itube import IFrame
+from tubes.tube import series, tube, receiver
+from tubes.framing import bytesToLines, linesToBytes
+from tubes.fan import Out, In
+from tubes.listening import Listener
+from tubes.protocol import flowFountFromEndpoint
 
 
 
@@ -130,12 +131,12 @@ class Hub(object):
         self.participants = []
         self.channels = defaultdict(Channel)
 
-    def newParticipantFlow(self, fount, drain):
-        commandFount = fount.flowTo(
+    def newParticipantFlow(self, flow):
+        commandFount = flow.fount.flowTo(
             series(OnStop(lambda: self.participants.remove(participant)),
                    bytesToLines(), linesToCommands)
         )
-        commandDrain = series(commandsToLines, linesToBytes(), drain)
+        commandDrain = series(commandsToLines, linesToBytes(), flow.drain)
         participant = Participant(self, commandFount, commandDrain)
         self.participants.append(participant)
 
@@ -144,10 +145,12 @@ class Hub(object):
 
 
 
+@inlineCallbacks
 def main(reactor, port="stdio:"):
     endpoint = serverFromString(reactor, port)
-    endpoint.listen(factoryFromFlow(Hub().newParticipantFlow))
-    return Deferred()
+    flowFount = yield flowFountFromEndpoint(endpoint)
+    flowFount.flowTo(Listener(Hub().newParticipantFlow))
+    yield Deferred()
 
 
 
diff --git a/sketches/notes.rst b/sketches/notes.rst
index 91e3587..7425177 100644
--- a/sketches/notes.rst
+++ b/sketches/notes.rst
@@ -3,18 +3,18 @@ In the interst of making this branch more accessible to additional contributors,
 Framing needs a ton of tests.
 It hasn't changed a whole lot so documenting and testing this module might be a good way to get started.
 
-``twisted.tubes.protocol`` is pretty well tested and roughly complete but could really use some docstrings, and improve the ones it has.
-See for example the docstring for factoryFromFlow.
+``tubes.protocol`` is pretty well tested and roughly complete but could really use some docstrings, and improve the ones it has.
+See for example the docstring for flowFountFromEndpoint.
 
-The objects in ``twisted.tubes.protocol``, especially those that show up in log messages, could really use nicer reprs that indicate what they're doing.
+The objects in ``tubes.protocol``, especially those that show up in log messages, could really use nicer reprs that indicate what they're doing.
 For example ``_ProtocolPlumbing`` and ``_FlowFactory`` should both include information about the flow function they're working on behalf of.
 
-Similarly, ``twisted.tubes.fan`` is a pretty rough sketch, although it's a bit less self-evident what is going on there since it's not fully implemented.
+Similarly, ``tubes.fan`` is a pretty rough sketch, although it's a bit less self-evident what is going on there since it's not fully implemented.
 (*Hopefully* it's straightforward, but let's not count on hope.)
 
 There are a bunch of un-covered `__repr__`s, probably.
 
-`twisted.tubes.tube.Diverter` could use some better docstrings, as could its helpers `_DrainingFount` and `_DrainingTube`.
+`tubes.tube.Diverter` could use some better docstrings, as could its helpers `_DrainingFount` and `_DrainingTube`.
 
 We need a decorator for a function so that this:
 
diff --git a/tox.ini b/tox.ini
index ac04142..e689da0 100644
--- a/tox.ini
+++ b/tox.ini
@@ -1,5 +1,5 @@
 [tox]
-envlist = py26, py27, pypy, docs, lint
+envlist = py26, py27, pypy, docs, lint, apidocs
 
 [testenv]
 deps =
diff --git a/tubes/_siphon.py b/tubes/_siphon.py
index ad4c542..665c901 100644
--- a/tubes/_siphon.py
+++ b/tubes/_siphon.py
@@ -10,8 +10,8 @@ from collections import deque
 
 from zope.interface import implementer
 
-from .itube import IPause, IDrain, IFount, ITube
-from .kit import Pauser, beginFlowingFrom, beginFlowingTo
+from .itube import IDrain, IFount, ITube
+from .kit import Pauser, beginFlowingFrom, beginFlowingTo, NoPause, OncePause
 from ._components import _registryAdapting
 
 from twisted.python.failure import Failure
@@ -214,10 +214,7 @@ class _SiphonFount(_SiphonPiece):
             siphon's tube.
         """
         result = beginFlowingTo(self, drain)
-        if self._siphon._pauseBecauseNoDrain:
-            pbnd = self._siphon._pauseBecauseNoDrain
-            self._siphon._pauseBecauseNoDrain = None
-            pbnd.unpause()
+        self._siphon._pauseBecauseNoDrain.maybeUnpause()
         self._siphon._unbufferIterator()
         return result
 
@@ -245,19 +242,6 @@ class _SiphonFount(_SiphonPiece):
 
 
 
-@implementer(IPause)
-class _PlaceholderPause(object):
-    """
-    L{IPause} provider that does nothing.
-    """
-
-    def unpause(self):
-        """
-        No-op.
-        """
-
-
-
 @implementer(IDrain)
 class _SiphonDrain(_SiphonPiece):
     """
@@ -295,7 +279,7 @@ class _SiphonDrain(_SiphonPiece):
             pbpc = self._siphon._pauseBecausePauseCalled
             self._siphon._pauseBecausePauseCalled = None
             if fount is None:
-                pauseFlow = _PlaceholderPause
+                pauseFlow = NoPause
             else:
                 pauseFlow = fount.pauseFlow
             self._siphon._pauseBecausePauseCalled = pauseFlow()
@@ -382,9 +366,9 @@ class _Siphon(object):
         self._everStarted = False
         self._unbuffering = False
         self._flowStoppingReason = None
-        self._pauseBecauseNoDrain = None
 
         self._tfount = _SiphonFount(self)
+        self._pauseBecauseNoDrain = OncePause(self._tfount._pauser)
         self._tdrain = _SiphonDrain(self)
         self._tube = tube
         self._pending = SiphonPendingValues()
@@ -433,9 +417,7 @@ class _Siphon(object):
             return
         self._pending.append(iter(iterableOrNot))
         if self._tfount.drain is None:
-            if self._pauseBecauseNoDrain is None:
-                self._pauseBecauseNoDrain = self._tfount.pauseFlow()
-
+            self._pauseBecauseNoDrain.pauseOnce()
         self._unbufferIterator()
 
 
diff --git a/tubes/itube.py b/tubes/itube.py
index 9656155..8bfd04a 100644
--- a/tubes/itube.py
+++ b/tubes/itube.py
@@ -9,8 +9,8 @@ Interfaces related to data flows.
 from zope.interface import Interface, Attribute
 
 if 0:
-    from zope.interface.interfaces import IInterface
-    IInterface
+    from zope.interface.interfaces import ISpecification
+    ISpecification
     from twisted.python.failure import Failure
     Failure
 
@@ -60,7 +60,7 @@ class IFount(Interface):
         """
         The type of output produced by this Fount.
 
-        This may be an L{IInterface} provider.
+        This may be an L{ISpecification} provider.
         """)
 
     drain = Attribute(
@@ -126,6 +126,8 @@ class IDrain(Interface):
     inputType = Attribute(
         """
         Similar to L{IFount.outputType}.
+
+        This is an L{ISpecification} provider.
         """)
 
     fount = Attribute(
@@ -175,7 +177,7 @@ class IDrain(Interface):
 
 class ITube(Interface):
     """
-    A tube translates input to output.
+    A tube transforms input into output.
 
     Look at this awesome ASCII art::
 
diff --git a/tubes/kit.py b/tubes/kit.py
index 7ac796f..ee73633 100644
--- a/tubes/kit.py
+++ b/tubes/kit.py
@@ -132,3 +132,46 @@ def beginFlowingFrom(drain, fount):
          (oldFount.drain is drain) ):
         oldFount.flowTo(None)
 
+
+
+@implementer(IPause)
+class NoPause(object):
+    """
+    A null implementation of L{IPause} that does nothing.
+    """
+
+    def unpause(self):
+        """
+        No-op.
+        """
+
+
+
+class OncePause(object):
+    """
+    Pause a pauser once, unpause it if necessary.
+    """
+    def __init__(self, pauser):
+        """
+        Create a L{OncePause} with the given L{Pauser}.
+        """
+        self._pauser = pauser
+        self._currentlyPaused = False
+
+
+    def pauseOnce(self):
+        """
+        If this L{OncePause} is not currently paused, pause its pauser.
+        """
+        if not self._currentlyPaused:
+            self._currentlyPaused = True
+            self._pause = self._pauser.pause()
+
+
+    def maybeUnpause(self):
+        """
+        If this L{OncePause} is currently paused, unpause it.
+        """
+        if self._currentlyPaused:
+            self._currentlyPaused = False
+            self._pause.unpause()
diff --git a/tubes/listening.py b/tubes/listening.py
new file mode 100644
index 0000000..5b2657a
--- /dev/null
+++ b/tubes/listening.py
@@ -0,0 +1,135 @@
+# -*- test-case-name: tubes.test.test_listening -*-
+# Copyright (c) Twisted Matrix Laboratories.
+# See LICENSE for details.
+"""
+Listening.
+"""
+
+from zope.interface import implementer, implementedBy
+
+from .itube import IDrain
+from .kit import beginFlowingFrom, NoPause
+from .tube import tube, series
+
+class Flow(object):
+    """
+    A L{Flow} is a combination of a Fount and a Drain, representing a
+    bi-directional communication channel such as a TCP connection.
+
+    @ivar fount: A fount.
+    @type fount: L{IFount}
+
+    @ivar drain: A drain.
+    @type drain: L{IDrain}
+    """
+
+    def __init__(self, fount, drain):
+        """
+        @param fount: Fount.
+        @type fount: L{IFount}
+
+        @param drain: Drain.
+        @type drain: L{IDrain}
+        """
+        self.fount = fount
+        self.drain = drain
+
+
+
+@implementer(IDrain)
+class Listener(object):
+    """
+    A L{Listener} is a drain that accepts L{Flow}s and sets them up.
+    """
+
+    inputType = implementedBy(Flow)
+
+    def __init__(self, flowConnector, maxConnections=100):
+        """
+        @param flowConnector: a 1-argument callable taking a L{Flow} and
+            returning nothing, which connects the flow.
+
+        @param maxConnections: The number of concurrent L{Flow} objects
+            to maintain active at once.
+        @type maxConnections: L{int}
+        """
+        self.fount = None
+        self._flowConnector = flowConnector
+        self._maxConnections = maxConnections
+        self._currentConnections = 0
+        self._paused = NoPause()
+
+
+    def flowingFrom(self, fount):
+        """
+        The flow has begun from the given L{fount} of L{Flow}s.
+
+        @param fount: A fount of flows.  One example of such a suitable fount
+            would be the return value of
+            L{tubes.protocol.flowFountFromEndpoint}.
+
+        @return: L{None}, since this is a "terminal" drain, where founts of
+            L{Flow} must end up in order for more new connections to be
+            established.
+        """
+        beginFlowingFrom(self, fount)
+
+
+    def receive(self, item):
+        """
+        Receive the given flow, applying backpressure if too many connections
+        are active.
+
+        @param item: The inbound L{Flow}.
+        """
+        self._currentConnections += 1
+        if self._currentConnections >= self._maxConnections:
+            self._paused = self.fount.pauseFlow()
+        def dec():
+            self._currentConnections -= 1
+            self._paused.unpause()
+            self._paused = NoPause()
+        self._flowConnector(Flow(item.fount.flowTo(series(_OnStop(dec))),
+                                 item.drain))
+
+
+    def flowStopped(self, reason):
+        """
+        No more L{Flow}s are incoming; nothing to do.
+
+        @param reason: the reason the flow stopped.
+        """
+
+
+
+@tube
+class _OnStop(object):
+    """
+    Call a callback when the flow stops.
+    """
+    def __init__(self, callback):
+        """
+        Call the given callback.
+        """
+        self.callback = callback
+
+
+    def received(self, item):
+        """
+        Pass through all received items.
+
+        @param item: An item being passed through (type unknown).
+        """
+        yield item
+
+
+    def stopped(self, reason):
+        """
+        Call the callback on stop.
+
+        @param reason: the reason that the flow stopped; ignored.
+
+        @return: no items.
+        """
+        self.callback()
+        return ()
diff --git a/tubes/protocol.py b/tubes/protocol.py
index 6577f4c..6d44259 100644
--- a/tubes/protocol.py
+++ b/tubes/protocol.py
@@ -5,28 +5,34 @@
 """
 Objects to connect L{real data <_Protocol>} to L{tubes}.
 
-@see: L{factoryFromFlow}
+@see: L{flowFountFromEndpoint}
 """
 
 __all__ = [
-    'factoryFromFlow',
+    'flowFountFromEndpoint',
+    'flowFromEndpoint',
 ]
 
-from zope.interface import implementer
+from zope.interface import implementer, implementedBy
 
-from .kit import Pauser, beginFlowingFrom, beginFlowingTo
-from .itube import IDrain, IFount, ISegment
+from .kit import Pauser, beginFlowingFrom, beginFlowingTo, OncePause
+from .itube import StopFlowCalled, IDrain, IFount, ISegment
+from .listening import Flow
 
-from twisted.internet.interfaces import IPushProducer
+from twisted.python.failure import Failure
+from twisted.internet.interfaces import IPushProducer, IListeningPort
 from twisted.internet.protocol import Protocol as _Protocol
 
 if 0:
     # Workaround for inability of pydoctor to resolve references.
     from twisted.internet.interfaces import (
-        IProtocol, ITransport, IConsumer, IProtocolFactory, IProducer)
-    IProtocol, ITransport, IConsumer, IProtocolFactory, IProducer
-    from twisted.python.failure import Failure
-    Failure
+        IProtocol, ITransport, IConsumer, IProtocolFactory, IProducer,
+        IStreamServerEndpoint
+    )
+    (IProtocol, ITransport, IConsumer, IProtocolFactory, IProducer,
+     IStreamServerEndpoint)
+    from twisted.internet.defer import Deferred
+    Deferred
 
 
 
@@ -205,7 +211,7 @@ class _ProtocolPlumbing(_Protocol):
     A L{_ProtocolPlumbing} implements L{IProtocol} to deliver all incoming data
     to the drain associated with its L{fount <IFount>}.
 
-    @ivar _flow: A flow function, as described in L{factoryFromFlow}.
+    @ivar _flow: A flow function, as described in L{_factoryFromFlow}.
     @type _flow: L{callable}
 
     @ivar _drain: The drain that is passed on to the application, created after
@@ -267,17 +273,17 @@ class _ProtocolPlumbing(_Protocol):
 
 
 
-def factoryFromFlow(flow):
+def _factoryFromFlow(flow):
     """
     Convert a flow function into an L{IProtocolFactory}.
 
     A "flow function" is a function which takes a L{fount <IFount>} and an
     L{drain <IDrain>}.
 
-    L{factoryFromFlow} takes such a function and creates an L{IProtocolFactory}
-    which, upon each new connection, provides the flow function with an
-    L{IFount} and an L{IDrain} representing the read end and the write end of
-    the incoming connection, respectively.
+    L{_factoryFromFlow} takes such a function and creates an
+    L{IProtocolFactory} which, upon each new connection, provides the flow
+    function with an L{IFount} and an L{IDrain} representing the read end and
+    the write end of the incoming connection, respectively.
 
     @param flow: a 2-argument callable, taking (fount, drain).
     @type flow: L{callable}
@@ -287,3 +293,134 @@ def factoryFromFlow(flow):
     """
     from twisted.internet.protocol import Factory
     return Factory.forProtocol(lambda: _ProtocolPlumbing(flow))
+
+
+
+@implementer(IFount)
+class _FountImpl(object):
+    """
+    Implementation of fount for listening port.
+    """
+
+    outputType = implementedBy(Flow)
+
+    def __init__(self, portObject, aFlowFunction, preListen):
+        """
+        Create a fount implementation from a provider of L{IPushProducer} and a
+        function that takes a fount and a drain.
+
+        @param portObject: the result of the L{Deferred} from
+            L{IStreamServerEndpoint.listen}
+        @type portObject: L{IListeningPort} and L{IPushProducer} provider
+            (probably; workarounds are in place for other cases)
+
+        @param aFlowFunction: a 2-argument callable, invoked when a connection
+            arrives, with a fount and drain.
+        @type aFlowFunction: L{callable}
+
+        @param preListen: the founts and drains accepted before the C{listen}
+            L{Deferred} has fired.  Because these might be arriving before this
+            L{_FountImpl} even I{exists}, this needs to be passed in.  That is
+            OK because L{_FountImpl} is very tightly coupled to
+            L{flowFountFromEndpoint}, which is the only thing that constructs
+            it.
+        @type preListen: L{list} of 2-L{tuple}s of C{(fount, drain)}
+        """
+        self.drain = None
+        self._preListen = preListen
+        self._pauser = Pauser(portObject.pauseProducing,
+                              portObject.resumeProducing)
+        self._noDrainPause = OncePause(self._pauser)
+        self._aFlowFunction = aFlowFunction
+        self._portObject = portObject
+        if preListen:
+            self._noDrainPause.pauseOnce()
+
+
+    def flowTo(self, drain):
+        """
+        Start flowing to the given drain.
+
+        @param drain: The drain to send flows to.
+
+        @return: the next fount in the chain.
+        """
+        result = beginFlowingTo(self, drain)
+        self._noDrainPause.maybeUnpause()
+        for f, d in self._preListen:
+            self._aFlowFunction(f, d)
+        return result
+
+
+    def pauseFlow(self):
+        """
+        Allow backpressure to build up in the listening socket; ask Twisted to
+        stop calling C{accept}.
+
+        @return: An L{IPause}.
+        """
+        return self._pauser.pause()
+
+
+    def stopFlow(self):
+        """
+        Stop the delivery of L{Flow} objects to this L{_FountImpl}'s drain, and
+        stop listening on the port represented by this fount.
+        """
+        self.drain.flowStopped(Failure(StopFlowCalled()))
+        self.drain = None
+        if IListeningPort.providedBy(self._portObject):
+            self._portObject.stopListening()
+
+
+
+def flowFountFromEndpoint(endpoint):
+    """
+    Listen on the given endpoint, and thereby create a L{fount <IFount>} which
+    outputs a new L{Flow} for each connection.
+
+    @note: L{IStreamServerEndpoint} formally specifies that its C{connect}
+        method returns a L{Deferred} that fires with an L{IListeningPort}.
+        However, L{IListeningPort} is insufficient to express the requisite
+        flow-control to implement a fount; so the C{endpoint} parameter must be
+        an extended endpoint whose C{listen} L{Deferred} fires with a provider
+        of both L{IListeningPort} and L{IPushProducer}.  Luckily, the
+        real-world implementations of L{IListeningPort} within Twisted are all
+        L{IPushProducer}s as well, so practically speaking you will not notice
+        this, but for testing it is important to know this is necessary.
+
+    @param endpoint: a server endpoint.
+    @type endpoint: L{IStreamServerEndpoint}
+
+    @return: a L{twisted.internet.defer.Deferred} that fires with a L{IFount}
+        whose C{outputType} is L{Flow}.
+    """
+    preListen = []
+    def listening(portObject):
+        listening.impl = _FountImpl(portObject, aFlowFunction, preListen)
+        return listening.impl
+    listening.impl = None
+    def aFlowFunction(fount, drain):
+        if listening.impl is None or listening.impl.drain is None:
+            preListen.append((fount, drain))
+            if listening.impl is not None:
+                listening.impl._pauseForNoDrain()
+        else:
+            listening.impl.drain.receive(Flow(fount, drain))
+    aFactory = _factoryFromFlow(aFlowFunction)
+    return endpoint.listen(aFactory).addCallback(listening)
+
+
+
+def flowFromEndpoint(endpoint):
+    """
+    Convert a client endpoint into a L{Deferred} that fires with a L{Flow}.
+
+    @param endpoint: a client endpoint that will be connected to, once.
+
+    @return: a L{Deferred} that fires with a L{Flow}.
+    """
+    def cb(fount, drain):
+        cb.result = Flow(fount, drain)
+    return (endpoint.connect(_factoryFromFlow(cb))
+            .addCallback(lambda whatever: cb.result))
diff --git a/tubes/tube.py b/tubes/tube.py
index df954ef..fe23b49 100644
--- a/tubes/tube.py
+++ b/tubes/tube.py
@@ -15,8 +15,9 @@ from twisted.python.components import proxyForInterface
 from twisted.python.failure import Failure
 
 from .itube import IDrain, ITube, IDivertable, IFount, StopFlowCalled
-from ._siphon import _tubeRegistry, _Siphon, _PlaceholderPause, skip
+from ._siphon import _tubeRegistry, _Siphon, skip
 from ._components import _registryActive
+from .kit import NoPause as _PlaceholderPause
 
 __all__ = [
     "Diverter",


Test patch:
diff --git a/tubes/test/test_listening.py b/tubes/test/test_listening.py
new file mode 100644
index 0000000..5b0b3e3
--- /dev/null
+++ b/tubes/test/test_listening.py
@@ -0,0 +1,56 @@
+# -*- test-case-name: tubes.test.test_listening -*-
+# Copyright (c) Twisted Matrix Laboratories.
+# See LICENSE for details.
+
+"""
+Tests for L{tubes.listening}.
+"""
+
+from unittest import TestCase
+
+from ..listening import Flow, Listener
+from ..memory import iteratorFount
+
+from .util import FakeDrain
+
+class ListeningTests(TestCase):
+    """
+    Test cases for listening.
+    """
+
+    def test_listenerCallsFlowConnector(self):
+        """
+        A L{Listener} is a drain which calls the function given to it to
+        connect a flow
+        """
+        drained = FakeDrain()
+        flow = Flow(iteratorFount([1, 2, 3]),
+                    drained)
+        flows = []
+        fi = iteratorFount([flow])
+        listener = Listener(flows.append)
+        fi.flowTo(listener)
+        self.assertEqual(len(flows), 1)
+        results = FakeDrain()
+        flows[0].fount.flowTo(results)
+        # The listener might need to (and in fact does) interpose a different
+        # value for 'fount' and 'drain' to add hooks to them.  We assert about
+        # the values passed through them.
+        self.assertEqual(results.received, [1, 2, 3])
+        iteratorFount([4, 5, 6]).flowTo(flows[0].drain)
+        self.assertEqual(drained.received, [4, 5, 6])
+
+
+    def test_listenerLimitsConcurrentConnections(self):
+        """
+        L{Listener} will pause its fount when too many connections are
+        received.
+        """
+        connectorCalled = []
+        listener = Listener(connectorCalled.append, maxConnections=3)
+        tenFlows = iteratorFount([Flow(iteratorFount([1, 2, 3]),
+                                       FakeDrain())
+                                  for each in range(10)])
+        tenFlows.flowTo(listener)
+        self.assertEqual(len(connectorCalled), 3)
+        connectorCalled[0].fount.flowTo(connectorCalled[0].drain)
diff --git a/tubes/test/test_protocol.py b/tubes/test/test_protocol.py
index 8c64452..9586f82 100644
--- a/tubes/test/test_protocol.py
+++ b/tubes/test/test_protocol.py
@@ -6,14 +6,20 @@
 Tests for L{tubes.protocol}.
 """
 
+from zope.interface import implementer
+
 from twisted.trial.unittest import SynchronousTestCase as TestCase
 
 from twisted.python.failure import Failure
+from twisted.test.proto_helpers import StringTransport
+from twisted.internet.interfaces import IStreamServerEndpoint
 
-from ..protocol import factoryFromFlow
+from ..protocol import flowFountFromEndpoint, flowFromEndpoint
 from ..tube import tube, series
+from ..listening import Flow, Listener
+from ..itube import IFount
 
-from ..test.util import StringEndpoint, FakeDrain, FakeFount
+from .util import StringEndpoint, FakeDrain, FakeFount, fakeEndpointWithPorts
 
 @tube
 class RememberingTube(object):
@@ -50,9 +56,9 @@ class RememberingTube(object):
 
 
 
-class FlowingAdapterTests(TestCase):
+class FlowConnectorTests(TestCase):
     """
-    Tests for L{factoryFromFlow} and the drain/fount/factory adapters it
+    Tests for L{flowFromEndpoint} and the drain/fount/factory adapters it
     constructs.
     """
 
@@ -61,17 +67,25 @@ class FlowingAdapterTests(TestCase):
         Sert up these tests.
         """
         self.endpoint = StringEndpoint()
-        def flowFunction(fount, drain):
-            self.adaptedDrain = drain
-            self.adaptedFount = fount
-        self.adaptedProtocol = self.successResultOf(
-            self.endpoint.connect(factoryFromFlow(flowFunction))
-        )
-
+        flow = self.successResultOf(flowFromEndpoint(self.endpoint))
+        self.adaptedDrain = flow.drain
+        self.adaptedFount = flow.fount
         self.tube = RememberingTube()
         self.drain = series(self.tube)
 
 
+    def adaptedProtocol(self):
+        """
+        Retrieve a protocol for testing with.
+
+        @return: the first protocol instance to have been created by making the
+            virtual outbound connection associated with the call to
+            L{flowFromEndpoint} performed in L{FlowConnectorTests.setUp}.
+        @rtype: L{IProtocol}
+        """
+        return self.endpoint.transports[0].protocol
+
+
     def test_flowToSetsDrain(self):
         """
         L{_ProtocolFount.flowTo} will set the C{drain} attribute of the
@@ -87,7 +101,7 @@ class FlowingAdapterTests(TestCase):
         L{_ProtocolFount.dataReceived} to invoke L{receive} on its drain.
         """
         self.adaptedFount.flowTo(self.drain)
-        self.adaptedProtocol.dataReceived("some data")
+        self.adaptedProtocol().dataReceived("some data")
         self.assertEqual(self.tube.items, ["some data"])
 
 
@@ -108,7 +122,7 @@ class FlowingAdapterTests(TestCase):
         """
         self.adaptedFount.flowTo(self.drain)
         self.adaptedFount.stopFlow()
-        self.assertEqual(self.adaptedProtocol.transport.disconnecting, True)
+        self.assertEqual(self.adaptedProtocol().transport.disconnecting, True)
         # The connection has not been closed yet; we *asked* the flow to stop,
         # but it may not have done.
         self.assertEqual(self.tube.wasStopped, False)
@@ -121,7 +135,7 @@ class FlowingAdapterTests(TestCase):
         """
         self.adaptedFount.flowTo(self.drain)
         self.adaptedDrain.flowStopped(Failure(ZeroDivisionError()))
-        self.assertEqual(self.adaptedProtocol.transport.disconnecting, True)
+        self.assertEqual(self.adaptedProtocol().transport.disconnecting, True)
         self.assertEqual(self.tube.wasStopped, False)
 
 
@@ -137,7 +151,7 @@ class FlowingAdapterTests(TestCase):
         class MyFunException(Exception):
             pass
         f = Failure(MyFunException())
-        self.adaptedProtocol.connectionLost(f)
+        self.adaptedProtocol().connectionLost(f)
         self.assertEqual(self.tube.wasStopped, True)
         self.assertIdentical(f, self.tube.reason)
 
@@ -150,7 +164,7 @@ class FlowingAdapterTests(TestCase):
         ff = FakeFount()
         ff.flowTo(self.adaptedDrain)
         self.assertEqual(ff.flowIsStopped, False)
-        self.adaptedProtocol.connectionLost(Failure(ZeroDivisionError))
+        self.adaptedProtocol().connectionLost(Failure(ZeroDivisionError))
         self.assertEqual(ff.flowIsStopped, True)
 
 
@@ -160,16 +174,16 @@ class FlowingAdapterTests(TestCase):
         L{_ProtocolFount} is flowing to anything, then it will pause the
         transport but only until the L{_ProtocolFount} is flowing to something.
         """
-        self.adaptedProtocol.dataReceived("hello, ")
-        self.assertEqual(self.adaptedProtocol.transport.producerState,
+        self.adaptedProtocol().dataReceived("hello, ")
+        self.assertEqual(self.adaptedProtocol().transport.producerState,
                           'paused')
         # It would be invalid to call dataReceived again in this state, so no
         # need to test that...
         fd = FakeDrain()
         self.adaptedFount.flowTo(fd)
-        self.assertEqual(self.adaptedProtocol.transport.producerState,
+        self.assertEqual(self.adaptedProtocol().transport.producerState,
                          'producing')
-        self.adaptedProtocol.dataReceived("world!")
+        self.adaptedProtocol().dataReceived("world!")
         self.assertEqual(fd.received, ["hello, ", "world!"])
 
 
@@ -181,7 +195,7 @@ class FlowingAdapterTests(TestCase):
         self.test_dataReceivedBeforeFlowing()
         fd2 = FakeDrain()
         self.adaptedFount.flowTo(fd2)
-        self.adaptedProtocol.dataReceived("hooray")
+        self.adaptedProtocol().dataReceived("hooray")
         self.assertEqual(fd2.received, ["hooray"])
 
 
@@ -201,7 +215,7 @@ class FlowingAdapterTests(TestCase):
         """
         fd = FakeDrain()
         self.adaptedFount.flowTo(fd)
-        self.adaptedProtocol.dataReceived("a")
+        self.adaptedProtocol().dataReceived("a")
         self.adaptedFount.flowTo(None)
         self.assertEqual(fd.fount, None)
         self.test_dataReceivedBeforeFlowing()
@@ -231,10 +245,10 @@ class FlowingAdapterTests(TestCase):
         self.assertEqual(ff.flowIsPaused, False)
         self.adaptedDrain.flowingFrom(ff)
         # The connection is too full!  Back off!
-        self.adaptedProtocol.transport.producer.pauseProducing()
+        self.adaptedProtocol().transport.producer.pauseProducing()
         self.assertEqual(ff.flowIsPaused, True)
         # All clear, start writing again.
-        self.adaptedProtocol.transport.producer.resumeProducing()
+        self.adaptedProtocol().transport.producer.resumeProducing()
         self.assertEqual(ff.flowIsPaused, False)
 
 
@@ -249,17 +263,17 @@ class FlowingAdapterTests(TestCase):
         producing = 'producing'
         paused = 'paused'
         # Sanity check.
-        self.assertEqual(self.adaptedProtocol.transport.producerState,
+        self.assertEqual(self.adaptedProtocol().transport.producerState,
                          producing)
         self.adaptedFount.flowTo(fd)
         # Steady as she goes.
-        self.assertEqual(self.adaptedProtocol.transport.producerState,
+        self.assertEqual(self.adaptedProtocol().transport.producerState,
                          producing)
         anPause = fd.fount.pauseFlow()
-        self.assertEqual(self.adaptedProtocol.transport.producerState,
+        self.assertEqual(self.adaptedProtocol().transport.producerState,
                          paused)
         anPause.unpause()
-        self.assertEqual(self.adaptedProtocol.transport.producerState,
+        self.assertEqual(self.adaptedProtocol().transport.producerState,
                          producing)
 
 
@@ -288,3 +302,99 @@ class FlowingAdapterTests(TestCase):
                 return another
         anotherOther = self.adaptedFount.flowTo(ReflowingFakeDrain())
         self.assertIdentical(another, anotherOther)
+
+
+
+class FlowListenerTests(TestCase):
+    """
+    Tests for L{flowFountFromEndpoint} and the fount adapter it constructs.
+    """
+
+    def test_fromEndpoint(self):
+        """
+        L{flowFountFromEndpoint} returns a L{Deferred} that fires when the
+        listening port is ready.
+        """
+        endpoint, ports = fakeEndpointWithPorts()
+        deferred = flowFountFromEndpoint(endpoint)
+        self.assertNoResult(deferred)
+        deferred.callback(None)
+        result = self.successResultOf(deferred)
+        self.assertTrue(IFount.providedBy(result))
+        self.assertEqual(result.outputType.implementedBy(Flow), True)
+
+
+    def test_oneConnectionAccepted(self):
+        """
+        When a connection comes in to a listening L{flowFountFromEndpoint}, the
+        L{Listener} that it's flowing to's callback is called.
+        """
+        endpoint, ports = fakeEndpointWithPorts()
+        deferred = flowFountFromEndpoint(endpoint)
+        self.assertNoResult(deferred)
+        deferred.callback(None)
+        result = self.successResultOf(deferred)
+        connected = []
+        result.flowTo(Listener(connected.append))
+        protocol = ports[0].factory.buildProtocol(None)
+        self.assertEqual(len(connected), 0)
+        protocol.makeConnection(StringTransport())
+        self.assertEqual(len(connected), 1)
+
+
+    def test_acceptBeforeActuallyListening(self):
+        """
+        Sometimes a connection is established reentrantly by C{listen}, without
+        waiting for the L{Deferred} returned to fire.  In this case the
+        connection will be buffered until said L{Deferred} fires.
+        """
+        immediateTransport = StringTransport()
+        subEndpoint, ports = fakeEndpointWithPorts()
+        @implementer(IStreamServerEndpoint)
+        class ImmediateFakeEndpoint(object):
+            def listen(self, factory):
+                protocol = factory.buildProtocol(None)
+                protocol.makeConnection(immediateTransport)
+                return subEndpoint.listen(factory)
+        endpoint = ImmediateFakeEndpoint()
+        deferred = flowFountFromEndpoint(endpoint)
+        deferred.callback(None)
+        fount = self.successResultOf(deferred)
+        connected = []
+        self.assertEqual(ports[0].currentlyProducing, False)
+        fount.flowTo(Listener(connected.append))
+        self.assertEqual(ports[0].currentlyProducing, True)
+        self.assertEqual(len(connected), 1)
+
+
+    def test_backpressure(self):
+        """
+        When the L{IFount} returned by L{flowFountFromEndpoint} is paused, it
+        removes its listening port from the reactor.  When resumed, it re-adds
+        it.
+        """
+        endpoint, ports = fakeEndpointWithPorts()
+        deferred = flowFountFromEndpoint(endpoint)
+        deferred.callback(None)
+        fount = self.successResultOf(deferred)
+        fount.flowTo(FakeDrain())
+        pause = fount.pauseFlow()
+        self.assertEqual(ports[0].currentlyProducing, False)
+        pause.unpause()
+        self.assertEqual(ports[0].currentlyProducing, True)
+
+
+    def test_stopping(self):
+        """
+        The L{IFount} returned by L{flowFountFromEndpoint} will stop listening
+        on the endpoint
+        """
+        endpoint, ports = fakeEndpointWithPorts()
+        deferred = flowFountFromEndpoint(endpoint)
+        deferred.callback(None)
+        fount = self.successResultOf(deferred)
+        fd = FakeDrain()
+        fount.flowTo(fd)
+        fount.stopFlow()
+        self.assertEqual(ports[0].listenStopping, True)
+        self.assertEqual(len(fd.stopped), 1)
diff --git a/tubes/test/util.py b/tubes/test/util.py
index 2c29c0e..dd36224 100644
--- a/tubes/test/util.py
+++ b/tubes/test/util.py
@@ -11,12 +11,15 @@ from zope.interface.verify import verifyClass
 
 from twisted.test.proto_helpers import StringTransport
 from twisted.internet.defer import succeed
-from twisted.internet.interfaces import IStreamClientEndpoint
+from twisted.internet.interfaces import (
+    IStreamClientEndpoint, IStreamServerEndpoint, IListeningPort, IPushProducer
+)
 
 from ..itube import IDrain, IFount, IDivertable
 from ..tube import tube
 from ..kit import Pauser, beginFlowingFrom, beginFlowingTo
 
+from twisted.internet.defer import Deferred
 
 @implementer(IStreamClientEndpoint)
 class StringEndpoint(object):
@@ -311,3 +314,123 @@ class NullTube(object):
 
 
 
+@implementer(IListeningPort, IPushProducer)
+class FakeListeningProducerPort(object):
+    """
+    This is a fake L{IListeningPort}, also implementing L{IPushProducer}, which
+    L{flowFountFromEndpoint} needs to make backpressure work.
+    """
+    def __init__(self, factory):
+        """
+        Create a L{FakeListeningProducerPort} with the given protocol
+        factory.
+        """
+        self.factory = factory
+        self.stopper = Deferred()
+        self.listenStopping = False
+        self.currentlyProducing = True
+
+
+    def pauseProducing(self):
+        """
+        Pause producing new connections.
+        """
+        self.currentlyProducing = False
+
+
+    def resumeProducing(self):
+        """
+        Resume producing new connections.
+        """
+        self.currentlyProducing = True
+
+
+    def startListening(self):
+        """
+        Start listening on this port.
+
+        @raise CannotListenError: If it cannot listen on this port (e.g., it is
+            a TCP port and it cannot bind to the required port number).
+        """
+
+
+    def stopListening(self):
+        """
+        Stop listening on this fake port.
+
+        @return: a L{Deferred} that should be fired when the test wants to
+            complete stopping listening.
+        """
+        self.listenStopping = True
+        return self.stopper
+
+
+    def stopProducing(self):
+        """
+        Stop producing more data.
+        """
+        self.stopListening()
+
+
+    def getHost(self):
+        """
+        Get the host that this port is listening for.
+
+        @return: An L{IAddress} provider.
+        """
+
+verifyClass(IListeningPort, FakeListeningProducerPort)
+verifyClass(IPushProducer, FakeListeningProducerPort)
+
+
+@implementer(IStreamServerEndpoint)
+class FakeEndpoint(object):
+    """
+    A fake implementation of L{IStreamServerEndpoint} with a L{Deferred} that
+    fires controllably.
+
+    @ivar _listening: deferreds that will fire with listening ports when their
+        C{.callback} is invoked (input to C{.callback} ignored); added to when
+        C{listen} is called.
+    @type _listening: L{list} of L{Deferred}
+
+    @ivar _ports: list of ports that have already started listening
+    @type _ports: L{list} of L{IListeningPort}
+    """
+    def __init__(self):
+        """
+        Create a L{FakeEndpoint}.
+        """
+        self._listening = []
+        self._ports = []
+
+
+    def listen(self, factory):
+        """
+        Listen with the given factory.
+
+        @param factory: The factory to use for future connections.
+
+        @return: a L{Deferred} that fires with a new listening port.
+        """
+        self._listening.append(Deferred())
+        def newListener(ignored):
+            result = FakeListeningProducerPort(factory)
+            self._ports.append(result)
+            return result
+        return self._listening[-1].addCallback(newListener)
+
+
+
+def fakeEndpointWithPorts():
+    """
+    Create a L{FakeEndpoint} and expose the list of ports that it uses.
+
+    @return: a fake endpoint and a list of the ports it has listened on
+    @rtype: a 2-tuple of C{(endpoint, ports)}, where C{ports} is a L{list} of
+        L{IListeningPort}.
+    """
+    self = FakeEndpoint()
+    return self, self._ports
+
+verifyClass(IStreamServerEndpoint, FakeEndpoint)