Commands received/sent over the network. Command can represent command received from the debugger, or one to be sent by daemon.
| 41 | |
| 42 | |
| 43 | class NetCommand(_BaseNetCommand): |
| 44 | """ |
| 45 | Commands received/sent over the network. |
| 46 | |
| 47 | Command can represent command received from the debugger, |
| 48 | or one to be sent by daemon. |
| 49 | """ |
| 50 | |
| 51 | next_seq = 0 # sequence numbers |
| 52 | |
| 53 | _showing_debug_info = 0 |
| 54 | _show_debug_info_lock = ForkSafeLock(rlock=True) |
| 55 | |
| 56 | _after_send = None |
| 57 | |
| 58 | def __init__(self, cmd_id, seq, text, is_json=False): |
| 59 | """ |
| 60 | If sequence is 0, new sequence will be generated (otherwise, this was the response |
| 61 | to a command from the client). |
| 62 | """ |
| 63 | protocol = get_protocol() |
| 64 | self.id = cmd_id |
| 65 | if seq == 0: |
| 66 | NetCommand.next_seq += 2 |
| 67 | seq = NetCommand.next_seq |
| 68 | |
| 69 | self.seq = seq |
| 70 | |
| 71 | if is_json: |
| 72 | if hasattr(text, "to_dict"): |
| 73 | as_dict = text.to_dict(update_ids_to_dap=True) |
| 74 | else: |
| 75 | assert isinstance(text, dict) |
| 76 | as_dict = text |
| 77 | as_dict["pydevd_cmd_id"] = cmd_id |
| 78 | as_dict["seq"] = seq |
| 79 | self.as_dict = as_dict |
| 80 | try: |
| 81 | text = json.dumps(as_dict) |
| 82 | except TypeError: |
| 83 | text = json.dumps(as_dict, default=str) |
| 84 | |
| 85 | assert isinstance(text, str) |
| 86 | |
| 87 | if DebugInfoHolder.DEBUG_TRACE_LEVEL >= 1: |
| 88 | self._show_debug_info(cmd_id, seq, text) |
| 89 | |
| 90 | if is_json: |
| 91 | msg = text |
| 92 | else: |
| 93 | if protocol not in (HTTP_PROTOCOL, HTTP_JSON_PROTOCOL): |
| 94 | encoded = quote(to_string(text), '/<>_=" \t') |
| 95 | msg = "%s\t%s\t%s\n" % (cmd_id, seq, encoded) |
| 96 | |
| 97 | else: |
| 98 | msg = "%s\t%s\t%s" % (cmd_id, seq, text) |
| 99 | |
| 100 | if isinstance(msg, str): |
no test coverage detected