1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
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 ()
|
