Module briar_wrapper.api_thread
Handles background thread for Api
Should not be called from outside briar_wrapper
.
Expand source code
# Copyright (c) 2020 Nico Alt
# SPDX-License-Identifier: AGPL-3.0-only
# License-Filename: LICENSE.md
"""
Handles background thread for `briar_wrapper.api.Api`
**Should not be called from outside `briar_wrapper`.**
"""
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):
"""
Initialize with `briar_wrapper.api.Api` instance `api` and
path to Briar Headless JAR `headless_jar`
"""
self._api = api
self._command = ["java", "-jar", headless_jar]
def is_running(self):
"""
Returns `True` if background thread is runnning
"""
return (self._process is not None) and (self._process.poll() is None)
def start(self):
"""
Starts background thread
"""
if self.is_running():
raise Exception("API already running")
self._process = Popen(self._command, stdin=PIPE,
stdout=PIPE, stderr=STDOUT)
def watch(self, callback):
"""
Watches startup of background thread and calls `callback` once finished
"""
watch_thread = Thread(target=self._watch_thread, args=(callback,),
daemon=True)
watch_thread.start()
def login(self, password):
"""
Actually logins to Briar API with `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):
"""
Actually registers new account at Briar API with 2-tuple `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):
"""
Stops background thread
"""
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)
Classes
class ApiThread (api, headless_jar)
-
Initialize with
Api
instanceapi
and path to Briar Headless JARheadless_jar
Expand source code
class ApiThread: _api = None _process = None def __init__(self, api, headless_jar): """ Initialize with `briar_wrapper.api.Api` instance `api` and path to Briar Headless JAR `headless_jar` """ self._api = api self._command = ["java", "-jar", headless_jar] def is_running(self): """ Returns `True` if background thread is runnning """ return (self._process is not None) and (self._process.poll() is None) def start(self): """ Starts background thread """ if self.is_running(): raise Exception("API already running") self._process = Popen(self._command, stdin=PIPE, stdout=PIPE, stderr=STDOUT) def watch(self, callback): """ Watches startup of background thread and calls `callback` once finished """ watch_thread = Thread(target=self._watch_thread, args=(callback,), daemon=True) watch_thread.start() def login(self, password): """ Actually logins to Briar API with `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): """ Actually registers new account at Briar API with 2-tuple `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): """ Stops background thread """ 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)
Methods
def is_running(self)
-
Returns
True
if background thread is runnningExpand source code
def is_running(self): """ Returns `True` if background thread is runnning """ return (self._process is not None) and (self._process.poll() is None)
def login(self, password)
-
Actually logins to Briar API with
password
Expand source code
def login(self, password): """ Actually logins to Briar API with `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)
-
Actually registers new account at Briar API with 2-tuple
credentials
Expand source code
def register(self, credentials): """ Actually registers new account at Briar API with 2-tuple `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 start(self)
-
Starts background thread
Expand source code
def start(self): """ Starts background thread """ if self.is_running(): raise Exception("API already running") self._process = Popen(self._command, stdin=PIPE, stdout=PIPE, stderr=STDOUT)
def stop(self)
-
Stops background thread
Expand source code
def stop(self): """ Stops background thread """ if not self.is_running(): raise Exception("Nothing to stop") self._process.terminate()
def watch(self, callback)
-
Watches startup of background thread and calls
callback
once finishedExpand source code
def watch(self, callback): """ Watches startup of background thread and calls `callback` once finished """ watch_thread = Thread(target=self._watch_thread, args=(callback,), daemon=True) watch_thread.start()