If sequence is 0, new sequence will be generated (otherwise, this was the response to a command from the client).
(self, cmd_id, seq, text, is_json=False)
| 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): |
| 101 | msg = msg.encode("utf-8") |
| 102 | |
| 103 | assert isinstance(msg, bytes) |
| 104 | as_bytes = msg |
| 105 | self._as_bytes = as_bytes |
| 106 | |
| 107 | def send(self, sock): |
| 108 | as_bytes = self._as_bytes |
nothing calls this directly
no test coverage detected