Coverage for Lib/asyncio/protocols.py: 98%

40 statements  

« prev     ^ index     » next       coverage.py v7.8.0, created at 2025-04-15 02:02 +0000

1"""Abstract Protocol base classes.""" 

2 

3__all__ = ( 

4 'BaseProtocol', 'Protocol', 'DatagramProtocol', 

5 'SubprocessProtocol', 'BufferedProtocol', 

6) 

7 

8 

9class BaseProtocol: 

10 """Common base class for protocol interfaces. 

11 

12 Usually user implements protocols that derived from BaseProtocol 

13 like Protocol or ProcessProtocol. 

14 

15 The only case when BaseProtocol should be implemented directly is 

16 write-only transport like write pipe 

17 """ 

18 

19 __slots__ = () 

20 

21 def connection_made(self, transport): 

22 """Called when a connection is made. 

23 

24 The argument is the transport representing the pipe connection. 

25 To receive data, wait for data_received() calls. 

26 When the connection is closed, connection_lost() is called. 

27 """ 

28 

29 def connection_lost(self, exc): 

30 """Called when the connection is lost or closed. 

31 

32 The argument is an exception object or None (the latter 

33 meaning a regular EOF is received or the connection was 

34 aborted or closed). 

35 """ 

36 

37 def pause_writing(self): 

38 """Called when the transport's buffer goes over the high-water mark. 

39 

40 Pause and resume calls are paired -- pause_writing() is called 

41 once when the buffer goes strictly over the high-water mark 

42 (even if subsequent writes increases the buffer size even 

43 more), and eventually resume_writing() is called once when the 

44 buffer size reaches the low-water mark. 

45 

46 Note that if the buffer size equals the high-water mark, 

47 pause_writing() is not called -- it must go strictly over. 

48 Conversely, resume_writing() is called when the buffer size is 

49 equal or lower than the low-water mark. These end conditions 

50 are important to ensure that things go as expected when either 

51 mark is zero. 

52 

53 NOTE: This is the only Protocol callback that is not called 

54 through EventLoop.call_soon() -- if it were, it would have no 

55 effect when it's most needed (when the app keeps writing 

56 without yielding until pause_writing() is called). 

57 """ 

58 

59 def resume_writing(self): 

60 """Called when the transport's buffer drains below the low-water mark. 

61 

62 See pause_writing() for details. 

63 """ 

64 

65 

66class Protocol(BaseProtocol): 

67 """Interface for stream protocol. 

68 

69 The user should implement this interface. They can inherit from 

70 this class but don't need to. The implementations here do 

71 nothing (they don't raise exceptions). 

72 

73 When the user wants to requests a transport, they pass a protocol 

74 factory to a utility function (e.g., EventLoop.create_connection()). 

75 

76 When the connection is made successfully, connection_made() is 

77 called with a suitable transport object. Then data_received() 

78 will be called 0 or more times with data (bytes) received from the 

79 transport; finally, connection_lost() will be called exactly once 

80 with either an exception object or None as an argument. 

81 

82 State machine of calls: 

83 

84 start -> CM [-> DR*] [-> ER?] -> CL -> end 

85 

86 * CM: connection_made() 

87 * DR: data_received() 

88 * ER: eof_received() 

89 * CL: connection_lost() 

90 """ 

91 

92 __slots__ = () 

93 

94 def data_received(self, data): 

95 """Called when some data is received. 

96 

97 The argument is a bytes object. 

98 """ 

99 

100 def eof_received(self): 

101 """Called when the other end calls write_eof() or equivalent. 

102 

103 If this returns a false value (including None), the transport 

104 will close itself. If it returns a true value, closing the 

105 transport is up to the protocol. 

106 """ 

107 

108 

109class BufferedProtocol(BaseProtocol): 

110 """Interface for stream protocol with manual buffer control. 

111 

112 Event methods, such as `create_server` and `create_connection`, 

113 accept factories that return protocols that implement this interface. 

114 

115 The idea of BufferedProtocol is that it allows to manually allocate 

116 and control the receive buffer. Event loops can then use the buffer 

117 provided by the protocol to avoid unnecessary data copies. This 

118 can result in noticeable performance improvement for protocols that 

119 receive big amounts of data. Sophisticated protocols can allocate 

120 the buffer only once at creation time. 

121 

122 State machine of calls: 

123 

124 start -> CM [-> GB [-> BU?]]* [-> ER?] -> CL -> end 

125 

126 * CM: connection_made() 

127 * GB: get_buffer() 

128 * BU: buffer_updated() 

129 * ER: eof_received() 

130 * CL: connection_lost() 

131 """ 

132 

133 __slots__ = () 

134 

135 def get_buffer(self, sizehint): 

136 """Called to allocate a new receive buffer. 

137 

138 *sizehint* is a recommended minimal size for the returned 

139 buffer. When set to -1, the buffer size can be arbitrary. 

140 

141 Must return an object that implements the 

142 :ref:`buffer protocol <bufferobjects>`. 

143 It is an error to return a zero-sized buffer. 

144 """ 

145 

146 def buffer_updated(self, nbytes): 

147 """Called when the buffer was updated with the received data. 

148 

149 *nbytes* is the total number of bytes that were written to 

150 the buffer. 

151 """ 

152 

153 def eof_received(self): 

154 """Called when the other end calls write_eof() or equivalent. 

155 

156 If this returns a false value (including None), the transport 

157 will close itself. If it returns a true value, closing the 

158 transport is up to the protocol. 

159 """ 

160 

161 

162class DatagramProtocol(BaseProtocol): 

163 """Interface for datagram protocol.""" 

164 

165 __slots__ = () 

166 

167 def datagram_received(self, data, addr): 

168 """Called when some datagram is received.""" 

169 

170 def error_received(self, exc): 

171 """Called when a send or receive operation raises an OSError. 

172 

173 (Other than BlockingIOError or InterruptedError.) 

174 """ 

175 

176 

177class SubprocessProtocol(BaseProtocol): 

178 """Interface for protocol for subprocess calls.""" 

179 

180 __slots__ = () 

181 

182 def pipe_data_received(self, fd, data): 

183 """Called when the subprocess writes data into stdout/stderr pipe. 

184 

185 fd is int file descriptor. 

186 data is bytes object. 

187 """ 

188 

189 def pipe_connection_lost(self, fd, exc): 

190 """Called when a file descriptor associated with the child process is 

191 closed. 

192 

193 fd is the int file descriptor that was closed. 

194 """ 

195 

196 def process_exited(self): 

197 """Called when subprocess has exited.""" 

198 

199 

200def _feed_data_to_buffered_proto(proto, data): 

201 data_len = len(data) 

202 while data_len: 202 ↛ exitline 202 didn't return from function '_feed_data_to_buffered_proto' because the condition on line 202 was always true

203 buf = proto.get_buffer(data_len) 

204 buf_len = len(buf) 

205 if not buf_len: 

206 raise RuntimeError('get_buffer() returned an empty buffer') 

207 

208 if buf_len >= data_len: 

209 buf[:data_len] = data 

210 proto.buffer_updated(data_len) 

211 return 

212 else: 

213 buf[:buf_len] = data[:buf_len] 

214 proto.buffer_updated(buf_len) 

215 data = data[buf_len:] 

216 data_len = len(data)