Function read_process

tests_python/debugger_unittest.py:372–391

(stream, buffer, debug_stream, stream_name, finish)

def read_process(stream, buffer, debug_stream, stream_name, finish):
    while True:
        line = stream.readline()
        if not line:
            break

        line = line.decode("utf-8", errors="replace")

        if SHOW_STDOUT:
            debug_stream.write(
                "%s: %s"
                % (
                    stream_name,
                    line,
                )
            )
        buffer.append(line)

        if finish[0]:
            return


def start_in_daemon_thread(target, args):