Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/asyncio/transports.py: 41%
111 statements
« prev ^ index » next coverage.py v7.0.1, created at 2022-12-25 06:11 +0000
1"""Abstract Transport class."""
3__all__ = (
4 'BaseTransport', 'ReadTransport', 'WriteTransport',
5 'Transport', 'DatagramTransport', 'SubprocessTransport',
6)
9class BaseTransport:
10 """Base class for transports."""
12 __slots__ = ('_extra',)
14 def __init__(self, extra=None):
15 if extra is None:
16 extra = {}
17 self._extra = extra
19 def get_extra_info(self, name, default=None):
20 """Get optional transport information."""
21 return self._extra.get(name, default)
23 def is_closing(self):
24 """Return True if the transport is closing or closed."""
25 raise NotImplementedError
27 def close(self):
28 """Close the transport.
30 Buffered data will be flushed asynchronously. No more data
31 will be received. After all buffered data is flushed, the
32 protocol's connection_lost() method will (eventually) called
33 with None as its argument.
34 """
35 raise NotImplementedError
37 def set_protocol(self, protocol):
38 """Set a new protocol."""
39 raise NotImplementedError
41 def get_protocol(self):
42 """Return the current protocol."""
43 raise NotImplementedError
46class ReadTransport(BaseTransport):
47 """Interface for read-only transports."""
49 __slots__ = ()
51 def is_reading(self):
52 """Return True if the transport is receiving."""
53 raise NotImplementedError
55 def pause_reading(self):
56 """Pause the receiving end.
58 No data will be passed to the protocol's data_received()
59 method until resume_reading() is called.
60 """
61 raise NotImplementedError
63 def resume_reading(self):
64 """Resume the receiving end.
66 Data received will once again be passed to the protocol's
67 data_received() method.
68 """
69 raise NotImplementedError
72class WriteTransport(BaseTransport):
73 """Interface for write-only transports."""
75 __slots__ = ()
77 def set_write_buffer_limits(self, high=None, low=None):
78 """Set the high- and low-water limits for write flow control.
80 These two values control when to call the protocol's
81 pause_writing() and resume_writing() methods. If specified,
82 the low-water limit must be less than or equal to the
83 high-water limit. Neither value can be negative.
85 The defaults are implementation-specific. If only the
86 high-water limit is given, the low-water limit defaults to an
87 implementation-specific value less than or equal to the
88 high-water limit. Setting high to zero forces low to zero as
89 well, and causes pause_writing() to be called whenever the
90 buffer becomes non-empty. Setting low to zero causes
91 resume_writing() to be called only once the buffer is empty.
92 Use of zero for either limit is generally sub-optimal as it
93 reduces opportunities for doing I/O and computation
94 concurrently.
95 """
96 raise NotImplementedError
98 def get_write_buffer_size(self):
99 """Return the current size of the write buffer."""
100 raise NotImplementedError
102 def write(self, data):
103 """Write some data bytes to the transport.
105 This does not block; it buffers the data and arranges for it
106 to be sent out asynchronously.
107 """
108 raise NotImplementedError
110 def writelines(self, list_of_data):
111 """Write a list (or any iterable) of data bytes to the transport.
113 The default implementation concatenates the arguments and
114 calls write() on the result.
115 """
116 data = b''.join(list_of_data)
117 self.write(data)
119 def write_eof(self):
120 """Close the write end after flushing buffered data.
122 (This is like typing ^D into a UNIX program reading from stdin.)
124 Data may still be received.
125 """
126 raise NotImplementedError
128 def can_write_eof(self):
129 """Return True if this transport supports write_eof(), False if not."""
130 raise NotImplementedError
132 def abort(self):
133 """Close the transport immediately.
135 Buffered data will be lost. No more data will be received.
136 The protocol's connection_lost() method will (eventually) be
137 called with None as its argument.
138 """
139 raise NotImplementedError
142class Transport(ReadTransport, WriteTransport):
143 """Interface representing a bidirectional transport.
145 There may be several implementations, but typically, the user does
146 not implement new transports; rather, the platform provides some
147 useful transports that are implemented using the platform's best
148 practices.
150 The user never instantiates a transport directly; they call a
151 utility function, passing it a protocol factory and other
152 information necessary to create the transport and protocol. (E.g.
153 EventLoop.create_connection() or EventLoop.create_server().)
155 The utility function will asynchronously create a transport and a
156 protocol and hook them up by calling the protocol's
157 connection_made() method, passing it the transport.
159 The implementation here raises NotImplemented for every method
160 except writelines(), which calls write() in a loop.
161 """
163 __slots__ = ()
166class DatagramTransport(BaseTransport):
167 """Interface for datagram (UDP) transports."""
169 __slots__ = ()
171 def sendto(self, data, addr=None):
172 """Send data to the transport.
174 This does not block; it buffers the data and arranges for it
175 to be sent out asynchronously.
176 addr is target socket address.
177 If addr is None use target address pointed on transport creation.
178 """
179 raise NotImplementedError
181 def abort(self):
182 """Close the transport immediately.
184 Buffered data will be lost. No more data will be received.
185 The protocol's connection_lost() method will (eventually) be
186 called with None as its argument.
187 """
188 raise NotImplementedError
191class SubprocessTransport(BaseTransport):
193 __slots__ = ()
195 def get_pid(self):
196 """Get subprocess id."""
197 raise NotImplementedError
199 def get_returncode(self):
200 """Get subprocess returncode.
202 See also
203 http://docs.python.org/3/library/subprocess#subprocess.Popen.returncode
204 """
205 raise NotImplementedError
207 def get_pipe_transport(self, fd):
208 """Get transport for pipe with number fd."""
209 raise NotImplementedError
211 def send_signal(self, signal):
212 """Send signal to subprocess.
214 See also:
215 docs.python.org/3/library/subprocess#subprocess.Popen.send_signal
216 """
217 raise NotImplementedError
219 def terminate(self):
220 """Stop the subprocess.
222 Alias for close() method.
224 On Posix OSs the method sends SIGTERM to the subprocess.
225 On Windows the Win32 API function TerminateProcess()
226 is called to stop the subprocess.
228 See also:
229 http://docs.python.org/3/library/subprocess#subprocess.Popen.terminate
230 """
231 raise NotImplementedError
233 def kill(self):
234 """Kill the subprocess.
236 On Posix OSs the function sends SIGKILL to the subprocess.
237 On Windows kill() is an alias for terminate().
239 See also:
240 http://docs.python.org/3/library/subprocess#subprocess.Popen.kill
241 """
242 raise NotImplementedError
245class _FlowControlMixin(Transport):
246 """All the logic for (write) flow control in a mix-in base class.
248 The subclass must implement get_write_buffer_size(). It must call
249 _maybe_pause_protocol() whenever the write buffer size increases,
250 and _maybe_resume_protocol() whenever it decreases. It may also
251 override set_write_buffer_limits() (e.g. to specify different
252 defaults).
254 The subclass constructor must call super().__init__(extra). This
255 will call set_write_buffer_limits().
257 The user may call set_write_buffer_limits() and
258 get_write_buffer_size(), and their protocol's pause_writing() and
259 resume_writing() may be called.
260 """
262 __slots__ = ('_loop', '_protocol_paused', '_high_water', '_low_water')
264 def __init__(self, extra=None, loop=None):
265 super().__init__(extra)
266 assert loop is not None
267 self._loop = loop
268 self._protocol_paused = False
269 self._set_write_buffer_limits()
271 def _maybe_pause_protocol(self):
272 size = self.get_write_buffer_size()
273 if size <= self._high_water:
274 return
275 if not self._protocol_paused:
276 self._protocol_paused = True
277 try:
278 self._protocol.pause_writing()
279 except (SystemExit, KeyboardInterrupt):
280 raise
281 except BaseException as exc:
282 self._loop.call_exception_handler({
283 'message': 'protocol.pause_writing() failed',
284 'exception': exc,
285 'transport': self,
286 'protocol': self._protocol,
287 })
289 def _maybe_resume_protocol(self):
290 if (self._protocol_paused and
291 self.get_write_buffer_size() <= self._low_water):
292 self._protocol_paused = False
293 try:
294 self._protocol.resume_writing()
295 except (SystemExit, KeyboardInterrupt):
296 raise
297 except BaseException as exc:
298 self._loop.call_exception_handler({
299 'message': 'protocol.resume_writing() failed',
300 'exception': exc,
301 'transport': self,
302 'protocol': self._protocol,
303 })
305 def get_write_buffer_limits(self):
306 return (self._low_water, self._high_water)
308 def _set_write_buffer_limits(self, high=None, low=None):
309 if high is None:
310 if low is None:
311 high = 64 * 1024
312 else:
313 high = 4 * low
314 if low is None:
315 low = high // 4
317 if not high >= low >= 0:
318 raise ValueError(
319 f'high ({high!r}) must be >= low ({low!r}) must be >= 0')
321 self._high_water = high
322 self._low_water = low
324 def set_write_buffer_limits(self, high=None, low=None):
325 self._set_write_buffer_limits(high=high, low=low)
326 self._maybe_pause_protocol()
328 def get_write_buffer_size(self):
329 raise NotImplementedError