cengal.io.asock_io.versions.v_1.base
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 18import errno 19import socket 20 21from cengal.base.classes import BaseClassSettings 22from cengal.code_flow_control.smart_values.versions.v_0 import ResultExistence 23from cengal.data_containers.dynamic_list_of_pieces import \ 24 DynamicListOfPiecesDequeWithLengthControl 25from cengal.data_containers.fast_fifo import FIFODequeWithLengthControl 26from cengal.hardware.info.cpu.versions.v_0 import l2_cache_per_core 27 28from .abstract import * 29from .recv_buff_size_computer import RecvBuffSizeComputer 30 31""" 32Module Docstring 33Docstrings: http://www.python.org/dev/peps/pep-0257/ 34""" 35 36__author__ = "ButenkoMS <gtalk@butenkoms.space>" 37__copyright__ = "Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>" 38__credits__ = ["ButenkoMS <gtalk@butenkoms.space>", ] 39__license__ = "Apache License, Version 2.0" 40__version__ = "4.4.1" 41__maintainer__ = "ButenkoMS <gtalk@butenkoms.space>" 42__email__ = "gtalk@butenkoms.space" 43# __status__ = "Prototype" 44__status__ = "Development" 45# __status__ = "Production" 46 47 48SET_OF_CONNECTION_ERRORS = {errno.ECONNRESET, errno.ECONNREFUSED, errno.ECONNABORTED, errno.EPIPE, errno.ESHUTDOWN} 49INET_TYPE_CONNECTIONS = {socket.AF_INET, socket.AF_INET6} 50 51 52try: 53 BlockingIOError 54except NameError: 55 class BlockingIOError(OSError): 56 pass 57 58try: 59 InterruptedError 60except NameError: 61 class InterruptedError(OSError): 62 pass 63 64try: 65 ConnectionError 66except NameError: 67 class ConnectionError(OSError): 68 pass 69 70try: 71 BrokenPipeError 72except NameError: 73 class BrokenPipeError(ConnectionError): 74 pass 75 76try: 77 ConnectionAbortedError 78except NameError: 79 class ConnectionAbortedError(ConnectionError): 80 pass 81 82try: 83 ConnectionRefusedError 84except NameError: 85 class ConnectionRefusedError(ConnectionError): 86 pass 87 88try: 89 ConnectionResetError 90except NameError: 91 class ConnectionResetError(ConnectionError): 92 pass 93 94 95class InternalNotCriticalError(Exception): 96 pass 97 98 99MESSAGE_SIZE_LEN = 8 100SERVER_ANSWER__KEYWORD_ACCEPTED = b'OK' 101 102 103class SimpleNetworkError(Exception): 104 pass 105 106 107class ConnectionSettings(BaseClassSettings): 108 def __init__(self, 109 connection_type: ConnectionType, 110 socket_address=None, 111 keyword: bytes = None, 112 socket_family=socket.AF_INET, 113 socket_type=socket.SOCK_STREAM, 114 socket_protocol=0, 115 socket_fileno=None, 116 backlog=0, 117 non_socket_connection_settings: NonSocketConnectionSettings = None): 118 """ 119 :param connection_type: ConnectionType() 120 :param socket_address: './main.server.AF_UNIX.socket', ('localhost', 8080), ('::', 50007, 0, 0), , ect. 121 :param keyword: b'sdlkfj s894 saf 84ewksdhf sdf'. Can be None for a Super Server 122 :param socket_family: AF_INET (the default), AF_INET6, AF_UNIX, AF_CAN or AF_RDS 123 :param socket_type: SOCK_STREAM (the default), SOCK_DGRAM, SOCK_RAW or perhaps one of the other SOCK_ constants 124 :param socket_protocol: in the case where the address family is AF_CAN the protocol should be one of CAN_RAW or 125 CAN_BCM 126 :param socket_fileno: If fileno is specified, the other arguments are ignored, causing the socket with the 127 specified file descriptor to return 128 """ 129 self.connection_type = connection_type 130 self.keyword = keyword 131 self.socket_address = socket_address 132 self.socket_family = socket_family 133 self.socket_type = socket_type 134 self.socket_protocol = socket_protocol 135 self.socket_fileno = socket_fileno 136 self.backlog = backlog 137 self.non_socket_connection_settings = non_socket_connection_settings 138 self.expected_clients_with_empty_output_fifo = set() 139 140 141class IOCoreMemoryManagement: 142 def __init__(self): 143 self.global__data_size_limit = ResultExistence(True, 2 * 1024**3) 144 145 self.global_in__data_size_limit = ResultExistence(True, 512 * 1024**2) 146 self.global_in__data_full_size = ResultExistence(True, 0) 147 self.global_in__deletable_data_full_size = ResultExistence(True, 0) 148 149 self.global_out__data_size_limit = ResultExistence(True, 512 * 1024**2) 150 self.global_out__data_full_size = ResultExistence(True, 0) 151 self.global_out__deletable_data_full_size = ResultExistence(True, 0) 152 153 def link_to(self, parent): 154 self.global__data_size_limit = parent.global__data_size_limit 155 156 self.global_in__data_size_limit = parent.global_in__data_size_limit 157 self.global_in__data_full_size = parent.global_in__data_full_size 158 self.global_in__deletable_data_full_size = parent.global_in__deletable_data_full_size 159 160 self.global_out__data_size_limit = parent.global_out__data_size_limit 161 self.global_out__data_full_size = parent.global_out__data_full_size 162 self.global_out__deletable_data_full_size = parent.global_out__deletable_data_full_size 163 164 165class ASockIOCoreMemoryManagement(IOCoreMemoryManagement): 166 def __init__(self): 167 super(ASockIOCoreMemoryManagement, self).__init__() 168 169 self.socket_read_fixed_buffer_size = ResultExistence(True, 170 int(l2_cache_per_core() / 2) or 1024 ** 2) 171 # 1024**2 is the fastest fixed read buffer on my CPU. 172 # Also ingeneral, it should be the half of the CPU cache per core (UPD: I don't remember why. Maybe to save other memory to instructions when we are dealing with big amount of connections). 173 # 174 # My CPU is Intel Core i5 3570: 175 # Architecture x86-64 176 # Threads 4 threads 177 # L2 cache 1 MB 178 # L2 cache per core 0.25 MB/core 179 # L3 cache 6 MB 180 # L3 cache per core 1.5 MB/core 181 182 def link_to(self, parent): 183 super(ASockIOCoreMemoryManagement, self).link_to(parent) 184 try: 185 self.socket_read_fixed_buffer_size = parent.socket_read_fixed_buffer_size 186 except AttributeError: 187 pass 188 189 190class Connection: 191 def __init__(self, 192 connection_id=None, 193 connection_settings: ConnectionSettings = None, 194 connection__conn_addr: tuple = None, 195 connection_state: ConnectionState = None, 196 global_memory_management: ASockIOCoreMemoryManagement = None 197 ): 198 """ 199 200 :param connection_id: ID for this connection 201 :param connection__conn_addr: tuple(conn, addr) where conn is a socket, addr is an address 202 :param global_memory_management: global memory management obj 203 """ 204 self.id = connection_id 205 self.connection_settings = connection_settings 206 if connection__conn_addr is None: 207 self.conn = ResultExistence(False, None) 208 self.addr = ResultExistence(False, None) 209 else: 210 self.conn = ResultExistence(True, connection__conn_addr[0]) 211 self.addr = ResultExistence(True, connection__conn_addr[1]) 212 self.connection_state = connection_state 213 214 self.addr_info = None 215 self.host_names = None 216 217 self.recv_buff_size_computer = RecvBuffSizeComputer() 218 self.recv_buff_size = 0 219 self.calc_new_recv_buff_size(0) 220 221 self.should_be_closed = False # socket should be closed immediately. For example because of IO error. 222 self.ready_to_be_closed = False # socket should be closed, after all messages had been sent to client. 223 self.ready_for_deletion = False # connection should be deleted immediately. For example because of unexpected 224 # keyword. 225 226 self.keyword = None 227 228 self.current_input_memoryview = None 229 self.current_input_memoryview_offset = 0 230 self.current_input_memoryview_nbytes = 0 231 self.current_input_memoryview_diff = 0 232 self.current_input_memoryview_message_nbytes = 0 233 self.raw_input_from_client = DynamicListOfPiecesDequeWithLengthControl( 234 external_data_length=global_memory_management.global_in__data_full_size) 235 self.current_message_length = None # length of current input message (or None, if size waw not read yet) 236 self.input_from_client = FIFODequeWithLengthControl( 237 external_data_full_size=global_memory_management.global_in__data_full_size) 238 239 self.current_output_memoryview = None 240 self.output_to_client = FIFODequeWithLengthControl( 241 external_data_full_size=global_memory_management.global_out__data_full_size) 242 243 self.this_is_raw_connection = False 244 245 self.connected_expected_client_id = None 246 self.connected_expected_client = None 247 self.has_inline_processor = False 248 249 def calc_new_recv_buff_size(self, last_recv_amount): 250 self.recv_buff_size = self.recv_buff_size_computer.calc_new_recv_buff_size(last_recv_amount) 251 252 def remove(self): 253 self.raw_input_from_client.remove() 254 self.input_from_client.remove() 255 self.output_to_client.remove() 256 257 def add_data_to_output_buffer(self, data): 258 pass 259 260 def add_data_to_input_buffer(self, data): 261 pass 262 263 264class InlineProcessor(InlineWorkerBase): 265 __slots__ = ('client_id', 'keyword', 'socket_family', 'socket_type', 'socket_proto', 'addr_info', 'host_names', 266 'is_in_raw_mode', '__set__is_in_raw_mode', '__set__mark_socket_as_should_be_closed_immediately', 267 '__set__mark_socket_as_ready_to_be_closed', '__external_parameters_set_trigger', 'output_messages', 268 '__hold__client_id') 269 270 def __init__(self, client_id=None, keyword: bytes = None, socket_family=None, socket_type=None, socket_proto=None, 271 addr_info=None, host_names=None, external_parameters_set_trigger: Set = None): 272 """ 273 274 :param keyword: client keyword. You may check for a known keywords to act appropriately 275 :param socket_family: 276 :param socket_type: 277 :param socket_proto: 278 :param addr_info: result of socket.getaddrinfo() call 279 :param host_names: result of socket.gethostbyaddr() call 280 """ 281 super(InlineProcessor, self).__init__(client_id, keyword, socket_family, socket_type, socket_proto, 282 addr_info, host_names, external_parameters_set_trigger) 283 284 self.__hold__client_id = client_id 285 self.__set__is_in_raw_mode = False 286 self.__set__mark_socket_as_should_be_closed_immediately = False 287 self.__set__mark_socket_as_ready_to_be_closed = False 288 self.__external_parameters_set_trigger = external_parameters_set_trigger 289 290 def set__is_in_raw_mode(self, is_in_raw_mode: bool): 291 self.__set__is_in_raw_mode = is_in_raw_mode 292 self.__external_parameters_set_trigger.add(self.__hold__client_id) 293 294 def mark__socket_as_should_be_closed_immediately(self, mark_socket_as: bool): 295 self.__set__mark_socket_as_should_be_closed_immediately = mark_socket_as 296 self.__external_parameters_set_trigger.add(self.__hold__client_id) 297 298 def mark__socket_as_ready_to_be_closed(self, mark_socket_as: bool): 299 self.__set__mark_socket_as_ready_to_be_closed = mark_socket_as 300 self.__external_parameters_set_trigger.add(self.__hold__client_id) 301 302 def __getstate__(self): 303 return self.client_id, self.keyword, self.socket_family, self.socket_type, self.socket_proto, self.addr_info, \ 304 self.host_names, self.is_in_raw_mode, self.__hold__client_id, self.__set__is_in_raw_mode, \ 305 self.__set__mark_socket_as_should_be_closed_immediately, self.__set__mark_socket_as_ready_to_be_closed, \ 306 self.__external_parameters_set_trigger, self.output_messages 307 308 def __setstate__(self, state): 309 self.client_id, self.keyword, self.socket_family, self.socket_type, self.socket_proto, self.addr_info, \ 310 self.host_names, self.is_in_raw_mode, self.__hold__client_id, self.__set__is_in_raw_mode, \ 311 self.__set__mark_socket_as_should_be_closed_immediately, self.__set__mark_socket_as_ready_to_be_closed, \ 312 self.__external_parameters_set_trigger, self.output_messages = state 313 314 315class Client: 316 def __init__(self, connection_settings: ConnectionSettings, client_id=None, connection_id=None): 317 """ Dasdfd safd 318 319 :param client_id: ID of the expected client 320 :param connection_id: ID of the connection 321 :param connection_settings: useful ConnectionSettings parameters are {connection_type, keyword} - for a client, 322 and all - for the super server. 323 324 """ 325 326 self.id = client_id 327 self.connection_id = connection_id 328 # self.__connection = None # Нельзя! Потому что в этом случае, объект клиента станет несериализуемым 329 self.__connection = None # Можно! Сделаем сериализуемым через переопределение магических методов 330 self.connection_settings = connection_settings 331 self.connection_settings.check() 332 333 self.will_use_raw_client_connection = False 334 self.will_use_raw_connection_without_handshake = False 335 self.this_is_unknown_client = False 336 337 self.obj_for_inline_processing = None 338 339 def __getstate__(self): 340 # data_for_pickling = ( 341 # self.id, 342 # self.connection_id, 343 # self.connection_settings, 344 # self.will_use_raw_client_connection, 345 # self.will_use_raw_connection_without_handshake, 346 # self.this_is_unknown_client, 347 # self.obj_for_inline_processing 348 # ) 349 # return data_for_pickling 350 return self.id, \ 351 self.connection_id, \ 352 self.connection_settings, \ 353 self.will_use_raw_client_connection, \ 354 self.will_use_raw_connection_without_handshake, \ 355 self.this_is_unknown_client, \ 356 self.obj_for_inline_processing 357 358 def __setstate__(self, data_after_unpickling): 359 self.id, self.connection_id, self.connection_settings, self.will_use_raw_client_connection, \ 360 self.will_use_raw_connection_without_handshake, self.this_is_unknown_client, \ 361 self.obj_for_inline_processing = data_after_unpickling 362 363 self.__connection = None 364 365 366class CheckIsRawConnection: 367 def __call__(self, asock_io_core: 'ASockIOCore', connection_info: Connection) -> bool: 368 """ 369 :param asock_io_core: 370 :param connection_info: 371 :return: "True" if it is RAW connection for Unknow Client. "False" otherwise. 372 """ 373 result = False 374 try: 375 if connection_info.conn.result.family in {socket.AF_INET, socket.AF_INET6}: 376 if connection_info.addr.result[0] not in asock_io_core.set_of_gate_addresses: 377 # If connected not from local IP address 378 result = True 379 except: 380 pass 381 return result
SET_OF_CONNECTION_ERRORS =
{32, 103, 104, 108, 111}
INET_TYPE_CONNECTIONS =
{<AddressFamily.AF_INET: 2>, <AddressFamily.AF_INET6: 10>}
class
InternalNotCriticalError(builtins.Exception):
Common base class for all non-exit exceptions.
Inherited Members
- builtins.Exception
- Exception
- builtins.BaseException
- with_traceback
- args
MESSAGE_SIZE_LEN =
8
SERVER_ANSWER__KEYWORD_ACCEPTED =
b'OK'
class
SimpleNetworkError(builtins.Exception):
Common base class for all non-exit exceptions.
Inherited Members
- builtins.Exception
- Exception
- builtins.BaseException
- with_traceback
- args
class
ConnectionSettings(cengal.base.classes.versions.v_0.classes.BaseClassSettings):
108class ConnectionSettings(BaseClassSettings): 109 def __init__(self, 110 connection_type: ConnectionType, 111 socket_address=None, 112 keyword: bytes = None, 113 socket_family=socket.AF_INET, 114 socket_type=socket.SOCK_STREAM, 115 socket_protocol=0, 116 socket_fileno=None, 117 backlog=0, 118 non_socket_connection_settings: NonSocketConnectionSettings = None): 119 """ 120 :param connection_type: ConnectionType() 121 :param socket_address: './main.server.AF_UNIX.socket', ('localhost', 8080), ('::', 50007, 0, 0), , ect. 122 :param keyword: b'sdlkfj s894 saf 84ewksdhf sdf'. Can be None for a Super Server 123 :param socket_family: AF_INET (the default), AF_INET6, AF_UNIX, AF_CAN or AF_RDS 124 :param socket_type: SOCK_STREAM (the default), SOCK_DGRAM, SOCK_RAW or perhaps one of the other SOCK_ constants 125 :param socket_protocol: in the case where the address family is AF_CAN the protocol should be one of CAN_RAW or 126 CAN_BCM 127 :param socket_fileno: If fileno is specified, the other arguments are ignored, causing the socket with the 128 specified file descriptor to return 129 """ 130 self.connection_type = connection_type 131 self.keyword = keyword 132 self.socket_address = socket_address 133 self.socket_family = socket_family 134 self.socket_type = socket_type 135 self.socket_protocol = socket_protocol 136 self.socket_fileno = socket_fileno 137 self.backlog = backlog 138 self.non_socket_connection_settings = non_socket_connection_settings 139 self.expected_clients_with_empty_output_fifo = set()
ConnectionSettings( connection_type: cengal.io.asock_io.versions.v_1.abstract.ConnectionType, socket_address=None, keyword: bytes = None, socket_family=<AddressFamily.AF_INET: 2>, socket_type=<SocketKind.SOCK_STREAM: 1>, socket_protocol=0, socket_fileno=None, backlog=0, non_socket_connection_settings: cengal.io.asock_io.versions.v_1.abstract.NonSocketConnectionSettings = None)
109 def __init__(self, 110 connection_type: ConnectionType, 111 socket_address=None, 112 keyword: bytes = None, 113 socket_family=socket.AF_INET, 114 socket_type=socket.SOCK_STREAM, 115 socket_protocol=0, 116 socket_fileno=None, 117 backlog=0, 118 non_socket_connection_settings: NonSocketConnectionSettings = None): 119 """ 120 :param connection_type: ConnectionType() 121 :param socket_address: './main.server.AF_UNIX.socket', ('localhost', 8080), ('::', 50007, 0, 0), , ect. 122 :param keyword: b'sdlkfj s894 saf 84ewksdhf sdf'. Can be None for a Super Server 123 :param socket_family: AF_INET (the default), AF_INET6, AF_UNIX, AF_CAN or AF_RDS 124 :param socket_type: SOCK_STREAM (the default), SOCK_DGRAM, SOCK_RAW or perhaps one of the other SOCK_ constants 125 :param socket_protocol: in the case where the address family is AF_CAN the protocol should be one of CAN_RAW or 126 CAN_BCM 127 :param socket_fileno: If fileno is specified, the other arguments are ignored, causing the socket with the 128 specified file descriptor to return 129 """ 130 self.connection_type = connection_type 131 self.keyword = keyword 132 self.socket_address = socket_address 133 self.socket_family = socket_family 134 self.socket_type = socket_type 135 self.socket_protocol = socket_protocol 136 self.socket_fileno = socket_fileno 137 self.backlog = backlog 138 self.non_socket_connection_settings = non_socket_connection_settings 139 self.expected_clients_with_empty_output_fifo = set()
:param connection_type: ConnectionType() :param socket_address: './main.server.AF_UNIX.socket', ('localhost', 8080), ('::', 50007, 0, 0), , ect. :param keyword: b'sdlkfj s894 saf 84ewksdhf sdf'. Can be None for a Super Server :param socket_family: AF_INET (the default), AF_INET6, AF_UNIX, AF_CAN or AF_RDS :param socket_type: SOCK_STREAM (the default), SOCK_DGRAM, SOCK_RAW or perhaps one of the other SOCK_ constants :param socket_protocol: in the case where the address family is AF_CAN the protocol should be one of CAN_RAW or CAN_BCM :param socket_fileno: If fileno is specified, the other arguments are ignored, causing the socket with the specified file descriptor to return
Inherited Members
- cengal.base.classes.versions.v_0.classes.BaseClassSettings
- check
class
IOCoreMemoryManagement:
142class IOCoreMemoryManagement: 143 def __init__(self): 144 self.global__data_size_limit = ResultExistence(True, 2 * 1024**3) 145 146 self.global_in__data_size_limit = ResultExistence(True, 512 * 1024**2) 147 self.global_in__data_full_size = ResultExistence(True, 0) 148 self.global_in__deletable_data_full_size = ResultExistence(True, 0) 149 150 self.global_out__data_size_limit = ResultExistence(True, 512 * 1024**2) 151 self.global_out__data_full_size = ResultExistence(True, 0) 152 self.global_out__deletable_data_full_size = ResultExistence(True, 0) 153 154 def link_to(self, parent): 155 self.global__data_size_limit = parent.global__data_size_limit 156 157 self.global_in__data_size_limit = parent.global_in__data_size_limit 158 self.global_in__data_full_size = parent.global_in__data_full_size 159 self.global_in__deletable_data_full_size = parent.global_in__deletable_data_full_size 160 161 self.global_out__data_size_limit = parent.global_out__data_size_limit 162 self.global_out__data_full_size = parent.global_out__data_full_size 163 self.global_out__deletable_data_full_size = parent.global_out__deletable_data_full_size
def
link_to(self, parent):
154 def link_to(self, parent): 155 self.global__data_size_limit = parent.global__data_size_limit 156 157 self.global_in__data_size_limit = parent.global_in__data_size_limit 158 self.global_in__data_full_size = parent.global_in__data_full_size 159 self.global_in__deletable_data_full_size = parent.global_in__deletable_data_full_size 160 161 self.global_out__data_size_limit = parent.global_out__data_size_limit 162 self.global_out__data_full_size = parent.global_out__data_full_size 163 self.global_out__deletable_data_full_size = parent.global_out__deletable_data_full_size
166class ASockIOCoreMemoryManagement(IOCoreMemoryManagement): 167 def __init__(self): 168 super(ASockIOCoreMemoryManagement, self).__init__() 169 170 self.socket_read_fixed_buffer_size = ResultExistence(True, 171 int(l2_cache_per_core() / 2) or 1024 ** 2) 172 # 1024**2 is the fastest fixed read buffer on my CPU. 173 # Also ingeneral, it should be the half of the CPU cache per core (UPD: I don't remember why. Maybe to save other memory to instructions when we are dealing with big amount of connections). 174 # 175 # My CPU is Intel Core i5 3570: 176 # Architecture x86-64 177 # Threads 4 threads 178 # L2 cache 1 MB 179 # L2 cache per core 0.25 MB/core 180 # L3 cache 6 MB 181 # L3 cache per core 1.5 MB/core 182 183 def link_to(self, parent): 184 super(ASockIOCoreMemoryManagement, self).link_to(parent) 185 try: 186 self.socket_read_fixed_buffer_size = parent.socket_read_fixed_buffer_size 187 except AttributeError: 188 pass
class
Connection:
191class Connection: 192 def __init__(self, 193 connection_id=None, 194 connection_settings: ConnectionSettings = None, 195 connection__conn_addr: tuple = None, 196 connection_state: ConnectionState = None, 197 global_memory_management: ASockIOCoreMemoryManagement = None 198 ): 199 """ 200 201 :param connection_id: ID for this connection 202 :param connection__conn_addr: tuple(conn, addr) where conn is a socket, addr is an address 203 :param global_memory_management: global memory management obj 204 """ 205 self.id = connection_id 206 self.connection_settings = connection_settings 207 if connection__conn_addr is None: 208 self.conn = ResultExistence(False, None) 209 self.addr = ResultExistence(False, None) 210 else: 211 self.conn = ResultExistence(True, connection__conn_addr[0]) 212 self.addr = ResultExistence(True, connection__conn_addr[1]) 213 self.connection_state = connection_state 214 215 self.addr_info = None 216 self.host_names = None 217 218 self.recv_buff_size_computer = RecvBuffSizeComputer() 219 self.recv_buff_size = 0 220 self.calc_new_recv_buff_size(0) 221 222 self.should_be_closed = False # socket should be closed immediately. For example because of IO error. 223 self.ready_to_be_closed = False # socket should be closed, after all messages had been sent to client. 224 self.ready_for_deletion = False # connection should be deleted immediately. For example because of unexpected 225 # keyword. 226 227 self.keyword = None 228 229 self.current_input_memoryview = None 230 self.current_input_memoryview_offset = 0 231 self.current_input_memoryview_nbytes = 0 232 self.current_input_memoryview_diff = 0 233 self.current_input_memoryview_message_nbytes = 0 234 self.raw_input_from_client = DynamicListOfPiecesDequeWithLengthControl( 235 external_data_length=global_memory_management.global_in__data_full_size) 236 self.current_message_length = None # length of current input message (or None, if size waw not read yet) 237 self.input_from_client = FIFODequeWithLengthControl( 238 external_data_full_size=global_memory_management.global_in__data_full_size) 239 240 self.current_output_memoryview = None 241 self.output_to_client = FIFODequeWithLengthControl( 242 external_data_full_size=global_memory_management.global_out__data_full_size) 243 244 self.this_is_raw_connection = False 245 246 self.connected_expected_client_id = None 247 self.connected_expected_client = None 248 self.has_inline_processor = False 249 250 def calc_new_recv_buff_size(self, last_recv_amount): 251 self.recv_buff_size = self.recv_buff_size_computer.calc_new_recv_buff_size(last_recv_amount) 252 253 def remove(self): 254 self.raw_input_from_client.remove() 255 self.input_from_client.remove() 256 self.output_to_client.remove() 257 258 def add_data_to_output_buffer(self, data): 259 pass 260 261 def add_data_to_input_buffer(self, data): 262 pass
Connection( connection_id=None, connection_settings: ConnectionSettings = None, connection__conn_addr: tuple = None, connection_state: cengal.io.asock_io.versions.v_1.abstract.ConnectionState = None, global_memory_management: ASockIOCoreMemoryManagement = None)
192 def __init__(self, 193 connection_id=None, 194 connection_settings: ConnectionSettings = None, 195 connection__conn_addr: tuple = None, 196 connection_state: ConnectionState = None, 197 global_memory_management: ASockIOCoreMemoryManagement = None 198 ): 199 """ 200 201 :param connection_id: ID for this connection 202 :param connection__conn_addr: tuple(conn, addr) where conn is a socket, addr is an address 203 :param global_memory_management: global memory management obj 204 """ 205 self.id = connection_id 206 self.connection_settings = connection_settings 207 if connection__conn_addr is None: 208 self.conn = ResultExistence(False, None) 209 self.addr = ResultExistence(False, None) 210 else: 211 self.conn = ResultExistence(True, connection__conn_addr[0]) 212 self.addr = ResultExistence(True, connection__conn_addr[1]) 213 self.connection_state = connection_state 214 215 self.addr_info = None 216 self.host_names = None 217 218 self.recv_buff_size_computer = RecvBuffSizeComputer() 219 self.recv_buff_size = 0 220 self.calc_new_recv_buff_size(0) 221 222 self.should_be_closed = False # socket should be closed immediately. For example because of IO error. 223 self.ready_to_be_closed = False # socket should be closed, after all messages had been sent to client. 224 self.ready_for_deletion = False # connection should be deleted immediately. For example because of unexpected 225 # keyword. 226 227 self.keyword = None 228 229 self.current_input_memoryview = None 230 self.current_input_memoryview_offset = 0 231 self.current_input_memoryview_nbytes = 0 232 self.current_input_memoryview_diff = 0 233 self.current_input_memoryview_message_nbytes = 0 234 self.raw_input_from_client = DynamicListOfPiecesDequeWithLengthControl( 235 external_data_length=global_memory_management.global_in__data_full_size) 236 self.current_message_length = None # length of current input message (or None, if size waw not read yet) 237 self.input_from_client = FIFODequeWithLengthControl( 238 external_data_full_size=global_memory_management.global_in__data_full_size) 239 240 self.current_output_memoryview = None 241 self.output_to_client = FIFODequeWithLengthControl( 242 external_data_full_size=global_memory_management.global_out__data_full_size) 243 244 self.this_is_raw_connection = False 245 246 self.connected_expected_client_id = None 247 self.connected_expected_client = None 248 self.has_inline_processor = False
:param connection_id: ID for this connection :param connection__conn_addr: tuple(conn, addr) where conn is a socket, addr is an address :param global_memory_management: global memory management obj
class
InlineProcessor(cengal.io.asock_io.versions.v_1.abstract.InlineWorkerBase):
265class InlineProcessor(InlineWorkerBase): 266 __slots__ = ('client_id', 'keyword', 'socket_family', 'socket_type', 'socket_proto', 'addr_info', 'host_names', 267 'is_in_raw_mode', '__set__is_in_raw_mode', '__set__mark_socket_as_should_be_closed_immediately', 268 '__set__mark_socket_as_ready_to_be_closed', '__external_parameters_set_trigger', 'output_messages', 269 '__hold__client_id') 270 271 def __init__(self, client_id=None, keyword: bytes = None, socket_family=None, socket_type=None, socket_proto=None, 272 addr_info=None, host_names=None, external_parameters_set_trigger: Set = None): 273 """ 274 275 :param keyword: client keyword. You may check for a known keywords to act appropriately 276 :param socket_family: 277 :param socket_type: 278 :param socket_proto: 279 :param addr_info: result of socket.getaddrinfo() call 280 :param host_names: result of socket.gethostbyaddr() call 281 """ 282 super(InlineProcessor, self).__init__(client_id, keyword, socket_family, socket_type, socket_proto, 283 addr_info, host_names, external_parameters_set_trigger) 284 285 self.__hold__client_id = client_id 286 self.__set__is_in_raw_mode = False 287 self.__set__mark_socket_as_should_be_closed_immediately = False 288 self.__set__mark_socket_as_ready_to_be_closed = False 289 self.__external_parameters_set_trigger = external_parameters_set_trigger 290 291 def set__is_in_raw_mode(self, is_in_raw_mode: bool): 292 self.__set__is_in_raw_mode = is_in_raw_mode 293 self.__external_parameters_set_trigger.add(self.__hold__client_id) 294 295 def mark__socket_as_should_be_closed_immediately(self, mark_socket_as: bool): 296 self.__set__mark_socket_as_should_be_closed_immediately = mark_socket_as 297 self.__external_parameters_set_trigger.add(self.__hold__client_id) 298 299 def mark__socket_as_ready_to_be_closed(self, mark_socket_as: bool): 300 self.__set__mark_socket_as_ready_to_be_closed = mark_socket_as 301 self.__external_parameters_set_trigger.add(self.__hold__client_id) 302 303 def __getstate__(self): 304 return self.client_id, self.keyword, self.socket_family, self.socket_type, self.socket_proto, self.addr_info, \ 305 self.host_names, self.is_in_raw_mode, self.__hold__client_id, self.__set__is_in_raw_mode, \ 306 self.__set__mark_socket_as_should_be_closed_immediately, self.__set__mark_socket_as_ready_to_be_closed, \ 307 self.__external_parameters_set_trigger, self.output_messages 308 309 def __setstate__(self, state): 310 self.client_id, self.keyword, self.socket_family, self.socket_type, self.socket_proto, self.addr_info, \ 311 self.host_names, self.is_in_raw_mode, self.__hold__client_id, self.__set__is_in_raw_mode, \ 312 self.__set__mark_socket_as_should_be_closed_immediately, self.__set__mark_socket_as_ready_to_be_closed, \ 313 self.__external_parameters_set_trigger, self.output_messages = state
InlineProcessor( client_id=None, keyword: bytes = None, socket_family=None, socket_type=None, socket_proto=None, addr_info=None, host_names=None, external_parameters_set_trigger: Set = None)
271 def __init__(self, client_id=None, keyword: bytes = None, socket_family=None, socket_type=None, socket_proto=None, 272 addr_info=None, host_names=None, external_parameters_set_trigger: Set = None): 273 """ 274 275 :param keyword: client keyword. You may check for a known keywords to act appropriately 276 :param socket_family: 277 :param socket_type: 278 :param socket_proto: 279 :param addr_info: result of socket.getaddrinfo() call 280 :param host_names: result of socket.gethostbyaddr() call 281 """ 282 super(InlineProcessor, self).__init__(client_id, keyword, socket_family, socket_type, socket_proto, 283 addr_info, host_names, external_parameters_set_trigger) 284 285 self.__hold__client_id = client_id 286 self.__set__is_in_raw_mode = False 287 self.__set__mark_socket_as_should_be_closed_immediately = False 288 self.__set__mark_socket_as_ready_to_be_closed = False 289 self.__external_parameters_set_trigger = external_parameters_set_trigger
:param keyword: client keyword. You may check for a known keywords to act appropriately :param socket_family: :param socket_type: :param socket_proto: :param addr_info: result of socket.getaddrinfo() call :param host_names: result of socket.gethostbyaddr() call
Inherited Members
- cengal.io.asock_io.versions.v_1.abstract.InlineWorkerBase
- on__data_received
- on__output_buffers_are_empty
- on__connection_lost
class
Client:
316class Client: 317 def __init__(self, connection_settings: ConnectionSettings, client_id=None, connection_id=None): 318 """ Dasdfd safd 319 320 :param client_id: ID of the expected client 321 :param connection_id: ID of the connection 322 :param connection_settings: useful ConnectionSettings parameters are {connection_type, keyword} - for a client, 323 and all - for the super server. 324 325 """ 326 327 self.id = client_id 328 self.connection_id = connection_id 329 # self.__connection = None # Нельзя! Потому что в этом случае, объект клиента станет несериализуемым 330 self.__connection = None # Можно! Сделаем сериализуемым через переопределение магических методов 331 self.connection_settings = connection_settings 332 self.connection_settings.check() 333 334 self.will_use_raw_client_connection = False 335 self.will_use_raw_connection_without_handshake = False 336 self.this_is_unknown_client = False 337 338 self.obj_for_inline_processing = None 339 340 def __getstate__(self): 341 # data_for_pickling = ( 342 # self.id, 343 # self.connection_id, 344 # self.connection_settings, 345 # self.will_use_raw_client_connection, 346 # self.will_use_raw_connection_without_handshake, 347 # self.this_is_unknown_client, 348 # self.obj_for_inline_processing 349 # ) 350 # return data_for_pickling 351 return self.id, \ 352 self.connection_id, \ 353 self.connection_settings, \ 354 self.will_use_raw_client_connection, \ 355 self.will_use_raw_connection_without_handshake, \ 356 self.this_is_unknown_client, \ 357 self.obj_for_inline_processing 358 359 def __setstate__(self, data_after_unpickling): 360 self.id, self.connection_id, self.connection_settings, self.will_use_raw_client_connection, \ 361 self.will_use_raw_connection_without_handshake, self.this_is_unknown_client, \ 362 self.obj_for_inline_processing = data_after_unpickling 363 364 self.__connection = None
Client( connection_settings: ConnectionSettings, client_id=None, connection_id=None)
317 def __init__(self, connection_settings: ConnectionSettings, client_id=None, connection_id=None): 318 """ Dasdfd safd 319 320 :param client_id: ID of the expected client 321 :param connection_id: ID of the connection 322 :param connection_settings: useful ConnectionSettings parameters are {connection_type, keyword} - for a client, 323 and all - for the super server. 324 325 """ 326 327 self.id = client_id 328 self.connection_id = connection_id 329 # self.__connection = None # Нельзя! Потому что в этом случае, объект клиента станет несериализуемым 330 self.__connection = None # Можно! Сделаем сериализуемым через переопределение магических методов 331 self.connection_settings = connection_settings 332 self.connection_settings.check() 333 334 self.will_use_raw_client_connection = False 335 self.will_use_raw_connection_without_handshake = False 336 self.this_is_unknown_client = False 337 338 self.obj_for_inline_processing = None
Dasdfd safd
:param client_id: ID of the expected client :param connection_id: ID of the connection :param connection_settings: useful ConnectionSettings parameters are {connection_type, keyword} - for a client, and all - for the super server.
class
CheckIsRawConnection:
367class CheckIsRawConnection: 368 def __call__(self, asock_io_core: 'ASockIOCore', connection_info: Connection) -> bool: 369 """ 370 :param asock_io_core: 371 :param connection_info: 372 :return: "True" if it is RAW connection for Unknow Client. "False" otherwise. 373 """ 374 result = False 375 try: 376 if connection_info.conn.result.family in {socket.AF_INET, socket.AF_INET6}: 377 if connection_info.addr.result[0] not in asock_io_core.set_of_gate_addresses: 378 # If connected not from local IP address 379 result = True 380 except: 381 pass 382 return result