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
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
|
#!/usr/bin/env python3
import argparse
import json
import time
from threading import Thread
import dbus
import dbus.service
from session import Session
from typing import Union
class RWAService(dbus.service.Object):
def __init__(self, mockup_mode: bool):
self.mockup_mode = mockup_mode
self.bus = dbus.SessionBus()
name = dbus.service.BusName("org.ArcticaProject.RWA", bus=self.bus)
self.update_service_running = False
self.sessions = {}
super().__init__(name, "/RWA")
@dbus.service.method("org.ArcticaProject.RWA", out_signature="s")
def start(self):
"""Start a new remote session."""
# Start session
session = Session(mockup_mode)
# Add session to sessions list
self.sessions[session.pid] = session
# Start session update service
self._ensure_update_service()
return json.dumps(session.client_meta)
@dbus.service.method("org.ArcticaProject.RWA", in_signature="i", out_signature="s")
def status(self, pid: int) -> str:
"""Get status information about a service."""
return self._get_status(pid)
@dbus.service.method("org.ArcticaProject.RWA", in_signature="i", out_signature="s")
def refresh_status(self, pid: int) -> str:
"""Get status information about a service and refresh status before."""
self._update_session(pid)
return self._get_status(pid)
@dbus.service.method("org.ArcticaProject.RWA", in_signature="i")
def stop(self, pid: int):
"""Stop a remote session."""
session = self.sessions[pid]
session.stop()
def _get_status(self, pid: int) -> str:
session = self.sessions[pid]
return json.dumps(session.status)
def _ensure_update_service(self):
"""Start session update thread if it isn't already running."""
if not self.update_service_running:
self.update_thread = Thread(target=self._update_sessions)
self.update_thread.start()
def _update_session(self, pid: int):
"""Update the status of a session."""
session = self.sessions[pid]
print(f"Session #{session.pid}")
# Check if VNC process is still running
running = session.vnc_process_running
if running:
print("Session is running")
else:
print("Session is dead.")
session.stop()
del self.sessions[session.pid]
def _update_sessions(self):
"""Go through all running sessions and update their status using ``_update_session``."""
while len(self.sessions.values()) > 0:
for session in list(self.sessions.values()):
self._update_session(session.pid)
time.sleep(2)
self.update_service_running = False
# TODO Probably kill daemon here (quit main loop)
def str2bool(v: Union[str, bool, int]) -> bool:
"""Return true or false if the given string can be interpreted as a boolean otherwise raise an exception."""
if isinstance(v, bool):
return v
if v.lower() in ('yes', 'true', 't', 'y', '1', 1):
return True
elif v.lower() in ('no', 'false', 'f', 'n', '0', 0):
return False
else:
raise argparse.ArgumentTypeError('Boolean value expected.')
if __name__ == "__main__":
parser = argparse.ArgumentParser(description='DBus session service for ' +
'ArcticaProject\'s ' +
'Remote Web App')
parser.add_argument("-m", "--mockup-mode", type=str2bool, nargs='?',
const=True, default=False,
help="Activate mockup mode. Act like the session " +
"service but don\'t do changes or call other " +
"parts of RWA.")
args = parser.parse_args()
mockup_mode = args.mockup_mode
if mockup_mode:
print("All API responses are faked and should NOT BE USED IN " +
"PRODUCTION!")
import dbus.mainloop.glib
from gi.repository import GLib
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
loop = GLib.MainLoop()
object = RWAService(mockup_mode)
loop.run()
|
