cengal.parallel_execution.asyncio.efficient_streams.versions.v_0.efficient_streams_abstract
Module Docstring Docstrings: http://www.python.org/dev/peps/pep-0257/
1#!/usr/bin/env python 2# coding=utf-8 3 4# Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space> 5# 6# Licensed under the Apache License, Version 2.0 (the "License"); 7# you may not use this file except in compliance with the License. 8# You may obtain a copy of the License at 9# 10# http://www.apache.org/licenses/LICENSE-2.0 11# 12# Unless required by applicable law or agreed to in writing, software 13# distributed under the License is distributed on an "AS IS" BASIS, 14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 15# See the License for the specific language governing permissions and 16# limitations under the License. 17 18 19""" 20Module Docstring 21Docstrings: http://www.python.org/dev/peps/pep-0257/ 22""" 23 24 25__author__ = "ButenkoMS <gtalk@butenkoms.space>" 26__copyright__ = "Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>" 27__credits__ = ["ButenkoMS <gtalk@butenkoms.space>", ] 28__license__ = "Apache License, Version 2.0" 29__version__ = "4.4.1" 30__maintainer__ = "ButenkoMS <gtalk@butenkoms.space>" 31__email__ = "gtalk@butenkoms.space" 32# __status__ = "Prototype" 33__status__ = "Development" 34# __status__ = "Production" 35 36 37__all__ = ['StreamManagerAbstract', 'StreamReaderAbstract', 'StreamReaderProtocolAbstract', 'StreamWriterAbstract'] 38 39 40from typing import Tuple, Set, Optional, Union 41from .efficient_streams_base_internal import * 42from .efficient_streams_base import * 43 44 45class StreamManagerAbstract: 46 def __init__(self) -> None: 47 raise NotImplementedError 48 49 async def open_connection(self, host=None, port=None, *, 50 loop=None, limit=DEFAULT_LIMIT, 51 stream_type: StreamType = StreamType.general, stream_name: str = str(), 52 protocol_greeting: Optional[str] = None, message_size_len: Optional[int] = None, 53 max_message_size_len: Optional[int] = None, 54 **kwds) -> Tuple['StreamReaderAbstract', 'StreamWriterAbstract']: 55 """A wrapper for create_connection() returning a (reader, writer) pair. 56 57 The reader returned is a StreamReaderAbstract instance; the writer is a 58 StreamWriterAbstract instance. 59 60 The arguments are all the usual arguments to create_connection() 61 except protocol_factory; most common are positional host and port, 62 with various optional keyword arguments following. 63 64 Additional optional keyword arguments are loop (to set the event loop 65 instance to use) and limit (to set the buffer limit passed to the 66 StreamReaderAbstract). 67 68 (If you want to customize the StreamReaderAbstract and/or 69 StreamReaderProtocolAbstract classes, just copy the code -- there's 70 really nothing special here except some convenience.) 71 """ 72 raise NotImplementedError 73 74 async def start_server(self, client_connected_cb, host=None, port=None, *, 75 loop=None, limit=DEFAULT_LIMIT, 76 stream_type: StreamType = StreamType.general, stream_name: str = str(), 77 gate_security_policy: GateSecurityPolicy = GateSecurityPolicy.disabled, policy_managed_stream_names: Optional[Set[str]] = None, 78 protocol_greeting: Optional[str] = None, message_size_len: Optional[int] = None, 79 max_message_size_len: Optional[int] = None, 80 **kwds): 81 """Start a socket server, call back for each client connected. 82 83 The first parameter, `client_connected_cb`, takes two parameters: 84 client_reader, client_writer. client_reader is a StreamReaderAbstract 85 object, while client_writer is a StreamWriterAbstract object. This 86 parameter can either be a plain callback function or a coroutine; 87 if it is a coroutine, it will be automatically converted into a 88 Task. 89 90 The rest of the arguments are all the usual arguments to 91 loop.create_server() except protocol_factory; most common are 92 positional host and port, with various optional keyword arguments 93 following. The return value is the same as loop.create_server(). 94 95 Additional optional keyword arguments are loop (to set the event loop 96 instance to use) and limit (to set the buffer limit passed to the 97 StreamReaderAbstract). 98 99 The return value is the same as loop.create_server(), i.e. a 100 Server object which can be used to stop the service. 101 """ 102 raise NotImplementedError 103 104 async def try_establish_message_protocol_server_side(self, reader: 'StreamReaderAbstract', writer: 'StreamWriterAbstract') -> bool: 105 raise NotImplementedError 106 107 async def try_establish_message_protocol_client_side(self, reader: 'StreamReaderAbstract', writer: 'StreamWriterAbstract') -> bool: 108 raise NotImplementedError 109 110 111class StreamReaderAbstract: 112 def __init__(self, manager: StreamManagerAbstract, message_protocol_settings: MessageProtocolSettings, *args, **kwargs) -> None: 113 raise NotImplementedError 114 115 async def read_max(self): 116 raise NotImplementedError 117 118 async def read_nearly_max(self): 119 raise NotImplementedError 120 121 async def read_with_counter(self): 122 raise NotImplementedError 123 124 def __repr__(self): 125 raise NotImplementedError 126 127 def at_eof(self): 128 raise NotImplementedError 129 130 async def readline(self): 131 """Read chunk of data from the stream until newline (b'\n') is found. 132 133 On success, return chunk that ends with newline. If only partial 134 line can be read due to EOF, return incomplete line without 135 terminating newline. When EOF was reached while no bytes read, empty 136 bytes object is returned. 137 138 If limit is reached, ValueError will be raised. In that case, if 139 newline was found, complete line including newline will be removed 140 from internal buffer. Else, internal buffer will be cleared. Limit is 141 compared against part of the line without newline. 142 143 If stream was paused, this function will automatically resume it if 144 needed. 145 """ 146 raise NotImplementedError 147 148 async def readuntil(self, separator=b'\n'): 149 """Read data from the stream until ``separator`` is found. 150 151 On success, the data and separator will be removed from the 152 internal buffer (consumed). Returned data will include the 153 separator at the end. 154 155 Configured stream limit is used to check result. Limit sets the 156 maximal length of data that can be returned, not counting the 157 separator. 158 159 If an EOF occurs and the complete separator is still not found, 160 an IncompleteReadError exception will be raised, and the internal 161 buffer will be reset. The IncompleteReadError.partial attribute 162 may contain the separator partially. 163 164 If the data cannot be read because of over limit, a 165 LimitOverrunError exception will be raised, and the data 166 will be left in the internal buffer, so it can be read again. 167 """ 168 raise NotImplementedError 169 170 async def read(self, n=-1): 171 """Read up to `n` bytes from the stream. 172 173 If n is not provided, or set to -1, read until EOF and return all read 174 bytes. If the EOF was received and the internal buffer is empty, return 175 an empty bytes object. 176 177 If n is zero, return empty bytes object immediately. 178 179 If n is positive, this function try to read `n` bytes, and may return 180 less or equal bytes than requested, but at least one byte. If EOF was 181 received before any byte is read, this function returns empty byte 182 object. 183 184 Returned value is not limited with limit, configured at stream 185 creation. 186 187 If stream was paused, this function will automatically resume it if 188 needed. 189 """ 190 raise NotImplementedError 191 192 async def read_nearly(self, n=-1): 193 """Read up to `n` bytes from the stream. 194 195 If n is not provided, or set to -1, read until EOF and return all read 196 bytes. If the EOF was received and the internal buffer is empty, return 197 an empty bytes object. 198 199 If n is zero, return empty bytes object immediately. 200 201 If n is positive, this function try to read `n` bytes, and may return 202 less or equal bytes than requested, but at least one byte. If EOF was 203 received before any byte is read, this function returns empty byte 204 object. 205 206 Returned value is not limited with limit, configured at stream 207 creation. 208 209 If stream was paused, this function will automatically resume it if 210 needed. 211 """ 212 raise NotImplementedError 213 214 async def readexactly(self, n): 215 """Read exactly `n` bytes. 216 217 Raise an IncompleteReadError if EOF is reached before `n` bytes can be 218 read. The IncompleteReadError.partial attribute of the exception will 219 contain the partial read bytes. 220 221 if n is zero, return empty bytes object. 222 223 Returned value is not limited with limit, configured at stream 224 creation. 225 226 If stream was paused, this function will automatically resume it if 227 needed. 228 """ 229 raise NotImplementedError 230 231 async def readonly_exactly(self, n): 232 raise NotImplementedError 233 234 async def read_message(self): 235 raise NotImplementedError 236 237 def message_awailable(self) -> bool: 238 raise NotImplementedError 239 240 def transport_pause_reading(self): 241 raise NotImplementedError 242 243 def transport_resume_reading(self): 244 raise NotImplementedError 245 246 247class StreamReaderProtocolAbstract: 248 def __init__(self, manager: StreamManagerAbstract, message_protocol_settings: MessageProtocolSettings, *args, **kwargs) -> None: 249 raise NotImplementedError 250 251 def connection_made(self, transport): 252 raise NotImplementedError 253 254 255class StreamWriterAbstract: 256 def __init__(self, *args, **kwargs) -> None: 257 raise NotImplementedError 258 259 def optimized_write(self, data): 260 raise NotImplementedError 261 262 def owrite(self, data): 263 raise NotImplementedError 264 265 async def partial_drain(self): 266 raise NotImplementedError 267 268 async def pdrain(self): 269 return await self.partial_drain() 270 271 async def full_drain(self): 272 raise NotImplementedError 273 274 async def fdrain(self): 275 return await self.full_drain() 276 277 def _release_autonomous_writer_waiters(self): 278 raise NotImplementedError 279 280 async def _autonomous_writer_impl(self): 281 raise NotImplementedError 282 283 def start_autonomous_writer(self): 284 raise NotImplementedError 285 286 def start_aw(self): 287 return self.start_autonomous_writer() 288 289 async def stop_autonomous_writer(self, timeout: Optional[Union[int, float]] = 0): 290 """_summary_ 291 292 Args: 293 timeout (Optional[Union[int, float]], optional): _description_. Defaults to 0. 0 - infinit timeout; None - default timeout 294 295 Returns: 296 _type_: _description_ 297 """ 298 raise NotImplementedError 299 300 async def stop_aw(self, timeout: Optional[Union[int, float]] = 0): 301 """_summary_ 302 303 Args: 304 timeout (Optional[Union[int, float]], optional): _description_. Defaults to 0. 0 - infinit timeout; None - default timeout 305 306 Returns: 307 _type_: _description_ 308 """ 309 return await self.stop_autonomous_writer(timeout) 310 311 async def autonomous_writer_drain_enough(self, lower_water_size: Optional[int] = None): 312 raise NotImplementedError 313 314 async def aw_drain_enough(self): 315 await self.autonomous_writer_drain_enough() 316 317 def optimized_write_message(self, data): 318 raise NotImplementedError 319 320 def owrite_message(self, data): 321 self.optimized_write_message(data) 322 323 async def send_message(self, data): 324 raise NotImplementedError
class
StreamManagerAbstract:
46class StreamManagerAbstract: 47 def __init__(self) -> None: 48 raise NotImplementedError 49 50 async def open_connection(self, host=None, port=None, *, 51 loop=None, limit=DEFAULT_LIMIT, 52 stream_type: StreamType = StreamType.general, stream_name: str = str(), 53 protocol_greeting: Optional[str] = None, message_size_len: Optional[int] = None, 54 max_message_size_len: Optional[int] = None, 55 **kwds) -> Tuple['StreamReaderAbstract', 'StreamWriterAbstract']: 56 """A wrapper for create_connection() returning a (reader, writer) pair. 57 58 The reader returned is a StreamReaderAbstract instance; the writer is a 59 StreamWriterAbstract instance. 60 61 The arguments are all the usual arguments to create_connection() 62 except protocol_factory; most common are positional host and port, 63 with various optional keyword arguments following. 64 65 Additional optional keyword arguments are loop (to set the event loop 66 instance to use) and limit (to set the buffer limit passed to the 67 StreamReaderAbstract). 68 69 (If you want to customize the StreamReaderAbstract and/or 70 StreamReaderProtocolAbstract classes, just copy the code -- there's 71 really nothing special here except some convenience.) 72 """ 73 raise NotImplementedError 74 75 async def start_server(self, client_connected_cb, host=None, port=None, *, 76 loop=None, limit=DEFAULT_LIMIT, 77 stream_type: StreamType = StreamType.general, stream_name: str = str(), 78 gate_security_policy: GateSecurityPolicy = GateSecurityPolicy.disabled, policy_managed_stream_names: Optional[Set[str]] = None, 79 protocol_greeting: Optional[str] = None, message_size_len: Optional[int] = None, 80 max_message_size_len: Optional[int] = None, 81 **kwds): 82 """Start a socket server, call back for each client connected. 83 84 The first parameter, `client_connected_cb`, takes two parameters: 85 client_reader, client_writer. client_reader is a StreamReaderAbstract 86 object, while client_writer is a StreamWriterAbstract object. This 87 parameter can either be a plain callback function or a coroutine; 88 if it is a coroutine, it will be automatically converted into a 89 Task. 90 91 The rest of the arguments are all the usual arguments to 92 loop.create_server() except protocol_factory; most common are 93 positional host and port, with various optional keyword arguments 94 following. The return value is the same as loop.create_server(). 95 96 Additional optional keyword arguments are loop (to set the event loop 97 instance to use) and limit (to set the buffer limit passed to the 98 StreamReaderAbstract). 99 100 The return value is the same as loop.create_server(), i.e. a 101 Server object which can be used to stop the service. 102 """ 103 raise NotImplementedError 104 105 async def try_establish_message_protocol_server_side(self, reader: 'StreamReaderAbstract', writer: 'StreamWriterAbstract') -> bool: 106 raise NotImplementedError 107 108 async def try_establish_message_protocol_client_side(self, reader: 'StreamReaderAbstract', writer: 'StreamWriterAbstract') -> bool: 109 raise NotImplementedError
async def
open_connection( self, host=None, port=None, *, loop=None, limit=10485760, stream_type: cengal.parallel_execution.asyncio.efficient_streams.versions.v_0.efficient_streams_base.StreamType = <StreamType.general: 0>, stream_name: str = '', protocol_greeting: Union[str, NoneType] = None, message_size_len: Union[int, NoneType] = None, max_message_size_len: Union[int, NoneType] = None, **kwds) -> tuple[StreamReaderAbstract, StreamWriterAbstract]:
50 async def open_connection(self, host=None, port=None, *, 51 loop=None, limit=DEFAULT_LIMIT, 52 stream_type: StreamType = StreamType.general, stream_name: str = str(), 53 protocol_greeting: Optional[str] = None, message_size_len: Optional[int] = None, 54 max_message_size_len: Optional[int] = None, 55 **kwds) -> Tuple['StreamReaderAbstract', 'StreamWriterAbstract']: 56 """A wrapper for create_connection() returning a (reader, writer) pair. 57 58 The reader returned is a StreamReaderAbstract instance; the writer is a 59 StreamWriterAbstract instance. 60 61 The arguments are all the usual arguments to create_connection() 62 except protocol_factory; most common are positional host and port, 63 with various optional keyword arguments following. 64 65 Additional optional keyword arguments are loop (to set the event loop 66 instance to use) and limit (to set the buffer limit passed to the 67 StreamReaderAbstract). 68 69 (If you want to customize the StreamReaderAbstract and/or 70 StreamReaderProtocolAbstract classes, just copy the code -- there's 71 really nothing special here except some convenience.) 72 """ 73 raise NotImplementedError
A wrapper for create_connection() returning a (reader, writer) pair.
The reader returned is a StreamReaderAbstract instance; the writer is a StreamWriterAbstract instance.
The arguments are all the usual arguments to create_connection() except protocol_factory; most common are positional host and port, with various optional keyword arguments following.
Additional optional keyword arguments are loop (to set the event loop instance to use) and limit (to set the buffer limit passed to the StreamReaderAbstract).
(If you want to customize the StreamReaderAbstract and/or StreamReaderProtocolAbstract classes, just copy the code -- there's really nothing special here except some convenience.)
async def
start_server( self, client_connected_cb, host=None, port=None, *, loop=None, limit=10485760, stream_type: cengal.parallel_execution.asyncio.efficient_streams.versions.v_0.efficient_streams_base.StreamType = <StreamType.general: 0>, stream_name: str = '', gate_security_policy: cengal.parallel_execution.asyncio.efficient_streams.versions.v_0.efficient_streams_base.GateSecurityPolicy = <GateSecurityPolicy.disabled: 1>, policy_managed_stream_names: Union[Set[str], NoneType] = None, protocol_greeting: Union[str, NoneType] = None, message_size_len: Union[int, NoneType] = None, max_message_size_len: Union[int, NoneType] = None, **kwds):
75 async def start_server(self, client_connected_cb, host=None, port=None, *, 76 loop=None, limit=DEFAULT_LIMIT, 77 stream_type: StreamType = StreamType.general, stream_name: str = str(), 78 gate_security_policy: GateSecurityPolicy = GateSecurityPolicy.disabled, policy_managed_stream_names: Optional[Set[str]] = None, 79 protocol_greeting: Optional[str] = None, message_size_len: Optional[int] = None, 80 max_message_size_len: Optional[int] = None, 81 **kwds): 82 """Start a socket server, call back for each client connected. 83 84 The first parameter, `client_connected_cb`, takes two parameters: 85 client_reader, client_writer. client_reader is a StreamReaderAbstract 86 object, while client_writer is a StreamWriterAbstract object. This 87 parameter can either be a plain callback function or a coroutine; 88 if it is a coroutine, it will be automatically converted into a 89 Task. 90 91 The rest of the arguments are all the usual arguments to 92 loop.create_server() except protocol_factory; most common are 93 positional host and port, with various optional keyword arguments 94 following. The return value is the same as loop.create_server(). 95 96 Additional optional keyword arguments are loop (to set the event loop 97 instance to use) and limit (to set the buffer limit passed to the 98 StreamReaderAbstract). 99 100 The return value is the same as loop.create_server(), i.e. a 101 Server object which can be used to stop the service. 102 """ 103 raise NotImplementedError
Start a socket server, call back for each client connected.
The first parameter, client_connected_cb, takes two parameters:
client_reader, client_writer. client_reader is a StreamReaderAbstract
object, while client_writer is a StreamWriterAbstract object. This
parameter can either be a plain callback function or a coroutine;
if it is a coroutine, it will be automatically converted into a
Task.
The rest of the arguments are all the usual arguments to loop.create_server() except protocol_factory; most common are positional host and port, with various optional keyword arguments following. The return value is the same as loop.create_server().
Additional optional keyword arguments are loop (to set the event loop instance to use) and limit (to set the buffer limit passed to the StreamReaderAbstract).
The return value is the same as loop.create_server(), i.e. a Server object which can be used to stop the service.
async def
try_establish_message_protocol_server_side( self, reader: StreamReaderAbstract, writer: StreamWriterAbstract) -> bool:
async def
try_establish_message_protocol_client_side( self, reader: StreamReaderAbstract, writer: StreamWriterAbstract) -> bool:
class
StreamReaderAbstract:
112class StreamReaderAbstract: 113 def __init__(self, manager: StreamManagerAbstract, message_protocol_settings: MessageProtocolSettings, *args, **kwargs) -> None: 114 raise NotImplementedError 115 116 async def read_max(self): 117 raise NotImplementedError 118 119 async def read_nearly_max(self): 120 raise NotImplementedError 121 122 async def read_with_counter(self): 123 raise NotImplementedError 124 125 def __repr__(self): 126 raise NotImplementedError 127 128 def at_eof(self): 129 raise NotImplementedError 130 131 async def readline(self): 132 """Read chunk of data from the stream until newline (b'\n') is found. 133 134 On success, return chunk that ends with newline. If only partial 135 line can be read due to EOF, return incomplete line without 136 terminating newline. When EOF was reached while no bytes read, empty 137 bytes object is returned. 138 139 If limit is reached, ValueError will be raised. In that case, if 140 newline was found, complete line including newline will be removed 141 from internal buffer. Else, internal buffer will be cleared. Limit is 142 compared against part of the line without newline. 143 144 If stream was paused, this function will automatically resume it if 145 needed. 146 """ 147 raise NotImplementedError 148 149 async def readuntil(self, separator=b'\n'): 150 """Read data from the stream until ``separator`` is found. 151 152 On success, the data and separator will be removed from the 153 internal buffer (consumed). Returned data will include the 154 separator at the end. 155 156 Configured stream limit is used to check result. Limit sets the 157 maximal length of data that can be returned, not counting the 158 separator. 159 160 If an EOF occurs and the complete separator is still not found, 161 an IncompleteReadError exception will be raised, and the internal 162 buffer will be reset. The IncompleteReadError.partial attribute 163 may contain the separator partially. 164 165 If the data cannot be read because of over limit, a 166 LimitOverrunError exception will be raised, and the data 167 will be left in the internal buffer, so it can be read again. 168 """ 169 raise NotImplementedError 170 171 async def read(self, n=-1): 172 """Read up to `n` bytes from the stream. 173 174 If n is not provided, or set to -1, read until EOF and return all read 175 bytes. If the EOF was received and the internal buffer is empty, return 176 an empty bytes object. 177 178 If n is zero, return empty bytes object immediately. 179 180 If n is positive, this function try to read `n` bytes, and may return 181 less or equal bytes than requested, but at least one byte. If EOF was 182 received before any byte is read, this function returns empty byte 183 object. 184 185 Returned value is not limited with limit, configured at stream 186 creation. 187 188 If stream was paused, this function will automatically resume it if 189 needed. 190 """ 191 raise NotImplementedError 192 193 async def read_nearly(self, n=-1): 194 """Read up to `n` bytes from the stream. 195 196 If n is not provided, or set to -1, read until EOF and return all read 197 bytes. If the EOF was received and the internal buffer is empty, return 198 an empty bytes object. 199 200 If n is zero, return empty bytes object immediately. 201 202 If n is positive, this function try to read `n` bytes, and may return 203 less or equal bytes than requested, but at least one byte. If EOF was 204 received before any byte is read, this function returns empty byte 205 object. 206 207 Returned value is not limited with limit, configured at stream 208 creation. 209 210 If stream was paused, this function will automatically resume it if 211 needed. 212 """ 213 raise NotImplementedError 214 215 async def readexactly(self, n): 216 """Read exactly `n` bytes. 217 218 Raise an IncompleteReadError if EOF is reached before `n` bytes can be 219 read. The IncompleteReadError.partial attribute of the exception will 220 contain the partial read bytes. 221 222 if n is zero, return empty bytes object. 223 224 Returned value is not limited with limit, configured at stream 225 creation. 226 227 If stream was paused, this function will automatically resume it if 228 needed. 229 """ 230 raise NotImplementedError 231 232 async def readonly_exactly(self, n): 233 raise NotImplementedError 234 235 async def read_message(self): 236 raise NotImplementedError 237 238 def message_awailable(self) -> bool: 239 raise NotImplementedError 240 241 def transport_pause_reading(self): 242 raise NotImplementedError 243 244 def transport_resume_reading(self): 245 raise NotImplementedError
StreamReaderAbstract( manager: StreamManagerAbstract, message_protocol_settings: cengal.parallel_execution.asyncio.efficient_streams.versions.v_0.efficient_streams_base_internal.MessageProtocolSettings, *args, **kwargs)
async def
readline(self):
131 async def readline(self): 132 """Read chunk of data from the stream until newline (b'\n') is found. 133 134 On success, return chunk that ends with newline. If only partial 135 line can be read due to EOF, return incomplete line without 136 terminating newline. When EOF was reached while no bytes read, empty 137 bytes object is returned. 138 139 If limit is reached, ValueError will be raised. In that case, if 140 newline was found, complete line including newline will be removed 141 from internal buffer. Else, internal buffer will be cleared. Limit is 142 compared against part of the line without newline. 143 144 If stream was paused, this function will automatically resume it if 145 needed. 146 """ 147 raise NotImplementedError
Read chunk of data from the stream until newline (b' ') is found.
On success, return chunk that ends with newline. If only partial
line can be read due to EOF, return incomplete line without
terminating newline. When EOF was reached while no bytes read, empty
bytes object is returned.
If limit is reached, ValueError will be raised. In that case, if
newline was found, complete line including newline will be removed
from internal buffer. Else, internal buffer will be cleared. Limit is
compared against part of the line without newline.
If stream was paused, this function will automatically resume it if
needed.
async def
readuntil(self, separator=b'\n'):
149 async def readuntil(self, separator=b'\n'): 150 """Read data from the stream until ``separator`` is found. 151 152 On success, the data and separator will be removed from the 153 internal buffer (consumed). Returned data will include the 154 separator at the end. 155 156 Configured stream limit is used to check result. Limit sets the 157 maximal length of data that can be returned, not counting the 158 separator. 159 160 If an EOF occurs and the complete separator is still not found, 161 an IncompleteReadError exception will be raised, and the internal 162 buffer will be reset. The IncompleteReadError.partial attribute 163 may contain the separator partially. 164 165 If the data cannot be read because of over limit, a 166 LimitOverrunError exception will be raised, and the data 167 will be left in the internal buffer, so it can be read again. 168 """ 169 raise NotImplementedError
Read data from the stream until separator is found.
On success, the data and separator will be removed from the internal buffer (consumed). Returned data will include the separator at the end.
Configured stream limit is used to check result. Limit sets the maximal length of data that can be returned, not counting the separator.
If an EOF occurs and the complete separator is still not found, an IncompleteReadError exception will be raised, and the internal buffer will be reset. The IncompleteReadError.partial attribute may contain the separator partially.
If the data cannot be read because of over limit, a LimitOverrunError exception will be raised, and the data will be left in the internal buffer, so it can be read again.
async def
read(self, n=-1):
171 async def read(self, n=-1): 172 """Read up to `n` bytes from the stream. 173 174 If n is not provided, or set to -1, read until EOF and return all read 175 bytes. If the EOF was received and the internal buffer is empty, return 176 an empty bytes object. 177 178 If n is zero, return empty bytes object immediately. 179 180 If n is positive, this function try to read `n` bytes, and may return 181 less or equal bytes than requested, but at least one byte. If EOF was 182 received before any byte is read, this function returns empty byte 183 object. 184 185 Returned value is not limited with limit, configured at stream 186 creation. 187 188 If stream was paused, this function will automatically resume it if 189 needed. 190 """ 191 raise NotImplementedError
Read up to n bytes from the stream.
If n is not provided, or set to -1, read until EOF and return all read bytes. If the EOF was received and the internal buffer is empty, return an empty bytes object.
If n is zero, return empty bytes object immediately.
If n is positive, this function try to read n bytes, and may return
less or equal bytes than requested, but at least one byte. If EOF was
received before any byte is read, this function returns empty byte
object.
Returned value is not limited with limit, configured at stream creation.
If stream was paused, this function will automatically resume it if needed.
async def
read_nearly(self, n=-1):
193 async def read_nearly(self, n=-1): 194 """Read up to `n` bytes from the stream. 195 196 If n is not provided, or set to -1, read until EOF and return all read 197 bytes. If the EOF was received and the internal buffer is empty, return 198 an empty bytes object. 199 200 If n is zero, return empty bytes object immediately. 201 202 If n is positive, this function try to read `n` bytes, and may return 203 less or equal bytes than requested, but at least one byte. If EOF was 204 received before any byte is read, this function returns empty byte 205 object. 206 207 Returned value is not limited with limit, configured at stream 208 creation. 209 210 If stream was paused, this function will automatically resume it if 211 needed. 212 """ 213 raise NotImplementedError
Read up to n bytes from the stream.
If n is not provided, or set to -1, read until EOF and return all read bytes. If the EOF was received and the internal buffer is empty, return an empty bytes object.
If n is zero, return empty bytes object immediately.
If n is positive, this function try to read n bytes, and may return
less or equal bytes than requested, but at least one byte. If EOF was
received before any byte is read, this function returns empty byte
object.
Returned value is not limited with limit, configured at stream creation.
If stream was paused, this function will automatically resume it if needed.
async def
readexactly(self, n):
215 async def readexactly(self, n): 216 """Read exactly `n` bytes. 217 218 Raise an IncompleteReadError if EOF is reached before `n` bytes can be 219 read. The IncompleteReadError.partial attribute of the exception will 220 contain the partial read bytes. 221 222 if n is zero, return empty bytes object. 223 224 Returned value is not limited with limit, configured at stream 225 creation. 226 227 If stream was paused, this function will automatically resume it if 228 needed. 229 """ 230 raise NotImplementedError
Read exactly n bytes.
Raise an IncompleteReadError if EOF is reached before n bytes can be
read. The IncompleteReadError.partial attribute of the exception will
contain the partial read bytes.
if n is zero, return empty bytes object.
Returned value is not limited with limit, configured at stream creation.
If stream was paused, this function will automatically resume it if needed.
class
StreamReaderProtocolAbstract:
248class StreamReaderProtocolAbstract: 249 def __init__(self, manager: StreamManagerAbstract, message_protocol_settings: MessageProtocolSettings, *args, **kwargs) -> None: 250 raise NotImplementedError 251 252 def connection_made(self, transport): 253 raise NotImplementedError
StreamReaderProtocolAbstract( manager: StreamManagerAbstract, message_protocol_settings: cengal.parallel_execution.asyncio.efficient_streams.versions.v_0.efficient_streams_base_internal.MessageProtocolSettings, *args, **kwargs)
class
StreamWriterAbstract:
256class StreamWriterAbstract: 257 def __init__(self, *args, **kwargs) -> None: 258 raise NotImplementedError 259 260 def optimized_write(self, data): 261 raise NotImplementedError 262 263 def owrite(self, data): 264 raise NotImplementedError 265 266 async def partial_drain(self): 267 raise NotImplementedError 268 269 async def pdrain(self): 270 return await self.partial_drain() 271 272 async def full_drain(self): 273 raise NotImplementedError 274 275 async def fdrain(self): 276 return await self.full_drain() 277 278 def _release_autonomous_writer_waiters(self): 279 raise NotImplementedError 280 281 async def _autonomous_writer_impl(self): 282 raise NotImplementedError 283 284 def start_autonomous_writer(self): 285 raise NotImplementedError 286 287 def start_aw(self): 288 return self.start_autonomous_writer() 289 290 async def stop_autonomous_writer(self, timeout: Optional[Union[int, float]] = 0): 291 """_summary_ 292 293 Args: 294 timeout (Optional[Union[int, float]], optional): _description_. Defaults to 0. 0 - infinit timeout; None - default timeout 295 296 Returns: 297 _type_: _description_ 298 """ 299 raise NotImplementedError 300 301 async def stop_aw(self, timeout: Optional[Union[int, float]] = 0): 302 """_summary_ 303 304 Args: 305 timeout (Optional[Union[int, float]], optional): _description_. Defaults to 0. 0 - infinit timeout; None - default timeout 306 307 Returns: 308 _type_: _description_ 309 """ 310 return await self.stop_autonomous_writer(timeout) 311 312 async def autonomous_writer_drain_enough(self, lower_water_size: Optional[int] = None): 313 raise NotImplementedError 314 315 async def aw_drain_enough(self): 316 await self.autonomous_writer_drain_enough() 317 318 def optimized_write_message(self, data): 319 raise NotImplementedError 320 321 def owrite_message(self, data): 322 self.optimized_write_message(data) 323 324 async def send_message(self, data): 325 raise NotImplementedError
async def
stop_autonomous_writer(self, timeout: Union[int, float, NoneType] = 0):
290 async def stop_autonomous_writer(self, timeout: Optional[Union[int, float]] = 0): 291 """_summary_ 292 293 Args: 294 timeout (Optional[Union[int, float]], optional): _description_. Defaults to 0. 0 - infinit timeout; None - default timeout 295 296 Returns: 297 _type_: _description_ 298 """ 299 raise NotImplementedError
_summary_
Args: timeout (Optional[Union[int, float]], optional): _description_. Defaults to 0. 0 - infinit timeout; None - default timeout
Returns: _type_: _description_
async def
stop_aw(self, timeout: Union[int, float, NoneType] = 0):
301 async def stop_aw(self, timeout: Optional[Union[int, float]] = 0): 302 """_summary_ 303 304 Args: 305 timeout (Optional[Union[int, float]], optional): _description_. Defaults to 0. 0 - infinit timeout; None - default timeout 306 307 Returns: 308 _type_: _description_ 309 """ 310 return await self.stop_autonomous_writer(timeout)
_summary_
Args: timeout (Optional[Union[int, float]], optional): _description_. Defaults to 0. 0 - infinit timeout; None - default timeout
Returns: _type_: _description_