| 385 | |
| 386 | # Proxy all socket data between the python code and the kubernetes websocket. |
| 387 | def _proxy(self): |
| 388 | channel_ports = [] |
| 389 | channel_initialized = [] |
| 390 | local_ports = {} |
| 391 | for port in self.local_ports.values(): |
| 392 | # Setup the data channel for this port number |
| 393 | channel_ports.append(port) |
| 394 | channel_initialized.append(False) |
| 395 | # Setup the error channel for this port number |
| 396 | channel_ports.append(port) |
| 397 | channel_initialized.append(False) |
| 398 | port.python.setblocking(True) |
| 399 | local_ports[port.python] = port |
| 400 | # The data to send on the websocket socket |
| 401 | kubernetes_data = b'' |
| 402 | while True: |
| 403 | rlist = [] # List of sockets to read from |
| 404 | wlist = [] # List of sockets to write to |
| 405 | if self.websocket.connected: |
| 406 | rlist.append(self.websocket) |
| 407 | if kubernetes_data: |
| 408 | wlist.append(self.websocket) |
| 409 | local_all_closed = True |
| 410 | for port in self.local_ports.values(): |
| 411 | if port.python.fileno() != -1: |
| 412 | if self.websocket.connected: |
| 413 | rlist.append(port.python) |
| 414 | if port.data: |
| 415 | wlist.append(port.python) |
| 416 | local_all_closed = False |
| 417 | else: |
| 418 | if port.data: |
| 419 | wlist.append(port.python) |
| 420 | local_all_closed = False |
| 421 | else: |
| 422 | port.python.close() |
| 423 | if local_all_closed and not (self.websocket.connected and kubernetes_data): |
| 424 | self.websocket.close() |
| 425 | return |
| 426 | r, w, _ = select.select(rlist, wlist, []) |
| 427 | for sock in r: |
| 428 | if sock == self.websocket: |
| 429 | pending = True |
| 430 | while pending: |
| 431 | try: |
| 432 | opcode, frame = self.websocket.recv_data_frame(True) |
| 433 | except WebSocketConnectionClosedException: |
| 434 | for port in self.local_ports.values(): |
| 435 | port.python.close() |
| 436 | return |
| 437 | if opcode == ABNF.OPCODE_BINARY: |
| 438 | if not frame.data: |
| 439 | raise RuntimeError("Unexpected frame data size") |
| 440 | channel = frame.data[0] |
| 441 | if channel >= len(channel_ports): |
| 442 | raise RuntimeError("Unexpected channel number: %s" % channel) |
| 443 | port = channel_ports[channel] |
| 444 | if channel_initialized[channel]: |