Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
# Copyright (c) 2020 Nico Alt
# SPDX-License-Identifier: AGPL-3.0-only
# License-Filename: LICENSE.md
from subprocess import Popen, PIPE, STDOUT
from threading import Thread
from time import sleep
from urllib.error import HTTPError, URLError
from urllib.request import urlopen
from briar_wrapper.constants import BASE_HTTP_URL
class ApiThread:
_api = None
_process = None
def __init__(self, api, headless_jar):
self._api = api
self._command = ["java", "-jar", headless_jar]
def is_running(self):
return (self._process is not None) and (self._process.poll() is None)
def start(self):
if self.is_running():
raise Exception("API already running")
self._process = Popen(self._command, stdin=PIPE,
stdout=PIPE, stderr=STDOUT)
def watch(self, callback):
watch_thread = Thread(target=self._watch_thread, args=(callback,),
daemon=True)
watch_thread.start()
def login(self, password):
if not self.is_running():
raise Exception("Can't login; API not running")
self._process.communicate((f"{password}\n").encode("utf-8"))
def register(self, credentials):
if not self.is_running():
raise Exception("Can't register; API not running")
self._process.communicate((credentials[0] + '\n' +
credentials[1] + '\n' +
credentials[1] + '\n').encode("utf-8"))
def stop(self):
if not self.is_running():
raise Exception("Nothing to stop")
self._process.terminate()
def _watch_thread(self, callback):
while self.is_running():
try:
urlopen(BASE_HTTP_URL)
sleep(0.1)
except HTTPError as http_error:
if http_error.code == 404:
return self._api.on_successful_startup(callback)
except URLError as url_error:
if not isinstance(url_error.reason, ConnectionRefusedError):
raise url_error
callback(False)