(self, session_id, repl_path, lean_env_path, initial_context=None,
timeout=600, expect_timeout=120)
| 18 | # Interactive thread class |
| 19 | class InteractiveThread(threading.Thread): |
| 20 | def __init__(self, session_id, repl_path, lean_env_path, initial_context=None, |
| 21 | timeout=600, expect_timeout=120): |
| 22 | super().__init__() |
| 23 | self.session_id = session_id |
| 24 | self.repl_path = repl_path |
| 25 | self.lean_env_path = lean_env_path |
| 26 | self.context = initial_context |
| 27 | self.session = None |
| 28 | self.expect_timeout = expect_timeout |
| 29 | |
| 30 | self.cmd_response_condition = threading.Event() |
| 31 | self.cmd_query_condition = threading.Event() |
| 32 | self.init_complete = threading.Event() |
| 33 | self.response = None |
| 34 | |
| 35 | self.stop_flag = False |
| 36 | self.timer = threading.Timer(timeout, self.stop) |
| 37 | |
| 38 | def initialize_check(self): |
| 39 | try: |
nothing calls this directly
no outgoing calls
no test coverage detected