Handle the Sec-WebSocket-Protocol HTTP response header. If provided, check that it contains exactly one supported subprotocol. Args: headers: WebSocket handshake response headers. Returns: Subprotocol, if one was selected.
(self, headers: Headers)
| 247 | return accepted_extensions |
| 248 | |
| 249 | def process_subprotocol(self, headers: Headers) -> Subprotocol | None: |
| 250 | """ |
| 251 | Handle the Sec-WebSocket-Protocol HTTP response header. |
| 252 | |
| 253 | If provided, check that it contains exactly one supported subprotocol. |
| 254 | |
| 255 | Args: |
| 256 | headers: WebSocket handshake response headers. |
| 257 | |
| 258 | Returns: |
| 259 | Subprotocol, if one was selected. |
| 260 | |
| 261 | """ |
| 262 | subprotocol: Subprotocol | None = None |
| 263 | |
| 264 | subprotocols = headers.get_all("Sec-WebSocket-Protocol") |
| 265 | |
| 266 | if subprotocols: |
| 267 | if self.available_subprotocols is None: |
| 268 | raise NegotiationError("no subprotocols supported") |
| 269 | |
| 270 | parsed_subprotocols: Sequence[Subprotocol] = sum( |
| 271 | [parse_subprotocol(header_value) for header_value in subprotocols], [] |
| 272 | ) |
| 273 | if len(parsed_subprotocols) > 1: |
| 274 | raise InvalidHeader( |
| 275 | "Sec-WebSocket-Protocol", |
| 276 | f"multiple values: {', '.join(parsed_subprotocols)}", |
| 277 | ) |
| 278 | |
| 279 | subprotocol = parsed_subprotocols[0] |
| 280 | if subprotocol not in self.available_subprotocols: |
| 281 | raise NegotiationError(f"unsupported subprotocol: {subprotocol}") |
| 282 | |
| 283 | return subprotocol |
| 284 | |
| 285 | def send_request(self, request: Request) -> None: |
| 286 | """ |
no test coverage detected