(conn: socket.socket,
stopped,
stopped_lock: threading.Lock)
| 6 | from ExploitProcessor import ExploitProcessor |
| 7 | |
| 8 | def send_loop(conn: socket.socket, |
| 9 | stopped, |
| 10 | stopped_lock: threading.Lock): |
| 11 | stopped_lock.acquire() |
| 12 | while not stopped[0]: |
| 13 | stopped_lock.release() |
| 14 | command = sys.stdin.readline() |
| 15 | stopped_lock.acquire() |
| 16 | if not stopped[0]: |
| 17 | conn.send(command.encode()) |
| 18 | |
| 19 | |
| 20 | """ |
nothing calls this directly
no outgoing calls
no test coverage detected