(
self,
request_callback: Union[
httputil.HTTPServerConnectionDelegate,
Callable[[httputil.HTTPServerRequest], None],
],
no_keep_alive: bool = False,
xheaders: bool = False,
ssl_options: Optional[Union[Dict[str, Any], ssl.SSLContext]] = None,
protocol: Optional[str] = None,
decompress_request: bool = False,
chunk_size: Optional[int] = None,
max_header_size: Optional[int] = None,
idle_connection_timeout: Optional[float] = None,
body_timeout: Optional[float] = None,
max_body_size: Optional[int] = None,
max_buffer_size: Optional[int] = None,
trusted_downstream: Optional[List[str]] = None,
)
| 162 | pass |
| 163 | |
| 164 | def initialize( |
| 165 | self, |
| 166 | request_callback: Union[ |
| 167 | httputil.HTTPServerConnectionDelegate, |
| 168 | Callable[[httputil.HTTPServerRequest], None], |
| 169 | ], |
| 170 | no_keep_alive: bool = False, |
| 171 | xheaders: bool = False, |
| 172 | ssl_options: Optional[Union[Dict[str, Any], ssl.SSLContext]] = None, |
| 173 | protocol: Optional[str] = None, |
| 174 | decompress_request: bool = False, |
| 175 | chunk_size: Optional[int] = None, |
| 176 | max_header_size: Optional[int] = None, |
| 177 | idle_connection_timeout: Optional[float] = None, |
| 178 | body_timeout: Optional[float] = None, |
| 179 | max_body_size: Optional[int] = None, |
| 180 | max_buffer_size: Optional[int] = None, |
| 181 | trusted_downstream: Optional[List[str]] = None, |
| 182 | ) -> None: |
| 183 | # This method's signature is not extracted with autodoc |
| 184 | # because we want its arguments to appear on the class |
| 185 | # constructor. When changing this signature, also update the |
| 186 | # copy in httpserver.rst. |
| 187 | self.request_callback = request_callback |
| 188 | self.xheaders = xheaders |
| 189 | self.protocol = protocol |
| 190 | self.conn_params = HTTP1ConnectionParameters( |
| 191 | decompress=decompress_request, |
| 192 | chunk_size=chunk_size, |
| 193 | max_header_size=max_header_size, |
| 194 | header_timeout=idle_connection_timeout or 3600, |
| 195 | max_body_size=max_body_size, |
| 196 | body_timeout=body_timeout, |
| 197 | no_keep_alive=no_keep_alive, |
| 198 | ) |
| 199 | TCPServer.__init__( |
| 200 | self, |
| 201 | ssl_options=ssl_options, |
| 202 | max_buffer_size=max_buffer_size, |
| 203 | read_chunk_size=chunk_size, |
| 204 | ) |
| 205 | self._connections = set() # type: Set[HTTP1ServerConnection] |
| 206 | self.trusted_downstream = trusted_downstream |
| 207 | |
| 208 | @classmethod |
| 209 | def configurable_base(cls) -> Type[Configurable]: |
nothing calls this directly
no test coverage detected