Converts a websocket agreed_parameters set to keyword arguments for our compressor objects.
(
self,
side: str,
agreed_parameters: Dict[str, Any],
compression_options: Optional[Dict[str, Any]] = None,
)
| 1002 | self.selected_subprotocol = headers.get("Sec-WebSocket-Protocol", None) |
| 1003 | |
| 1004 | def _get_compressor_options( |
| 1005 | self, |
| 1006 | side: str, |
| 1007 | agreed_parameters: Dict[str, Any], |
| 1008 | compression_options: Optional[Dict[str, Any]] = None, |
| 1009 | ) -> Dict[str, Any]: |
| 1010 | """Converts a websocket agreed_parameters set to keyword arguments |
| 1011 | for our compressor objects. |
| 1012 | """ |
| 1013 | options = dict( |
| 1014 | persistent=(side + "_no_context_takeover") not in agreed_parameters |
| 1015 | ) # type: Dict[str, Any] |
| 1016 | wbits_header = agreed_parameters.get(side + "_max_window_bits", None) |
| 1017 | if wbits_header is None: |
| 1018 | options["max_wbits"] = zlib.MAX_WBITS |
| 1019 | else: |
| 1020 | options["max_wbits"] = int(wbits_header) |
| 1021 | options["compression_options"] = compression_options |
| 1022 | return options |
| 1023 | |
| 1024 | def _create_compressors( |
| 1025 | self, |
no test coverage detected