Safely sends a response, handling pickle errors.
(self, req: RPCRequest,
response: RPCResponse)
| 640 | await self._send_response(req, response) |
| 641 | |
| 642 | async def _send_response(self, req: RPCRequest, |
| 643 | response: RPCResponse) -> bool: |
| 644 | """Safely sends a response, handling pickle errors.""" |
| 645 | try: |
| 646 | await self._client_socket.put_async(response, |
| 647 | routing_id=req.routing_id) |
| 648 | logger_debug(f"[server] Sent response for request {req.request_id}", |
| 649 | color="green") |
| 650 | return True |
| 651 | except Exception as e: |
| 652 | logger.error( |
| 653 | f"Failed to pickle response for request {req.request_id}: {e}") |
| 654 | error_msg = f"Failed to pickle response: {e}" |
| 655 | if req.is_streaming: |
| 656 | error_cls = RPCStreamingError |
| 657 | chunk_index = response.chunk_index if response else None |
| 658 | error_response = RPCResponse( |
| 659 | req.request_id, |
| 660 | result=None, |
| 661 | error=error_cls(error_msg, |
| 662 | traceback=traceback.format_exc()), |
| 663 | is_streaming=True, |
| 664 | chunk_index=chunk_index, |
| 665 | stream_status='error') |
| 666 | else: |
| 667 | error_cls = RPCError |
| 668 | error_response = RPCResponse( |
| 669 | req.request_id, |
| 670 | result=None, |
| 671 | error=error_cls(error_msg, |
| 672 | traceback=traceback.format_exc())) |
| 673 | |
| 674 | try: |
| 675 | await self._client_socket.put_async(error_response, |
| 676 | routing_id=req.routing_id) |
| 677 | logger_debug( |
| 678 | f"[server] Sent error response for request {req.request_id}", |
| 679 | color="green") |
| 680 | except Exception as e_inner: |
| 681 | logger.error( |
| 682 | f"Failed to send error response for request {req.request_id}: {e_inner}" |
| 683 | ) |
| 684 | return False |
| 685 | |
| 686 | def start(self) -> None: |
| 687 | """Binds sockets, starts workers, and begins proxying messages.""" |
no test coverage detected