summaryrefslogtreecommitdiff
path: root/voctocore/lib/controlserver.py
blob: bd263ff454f1784bea7c2b25c5fd0102f06fd447 (plain)
  1. #!/usr/bin/python3
  2. import socket, logging, traceback
  3. from queue import Queue
  4. from gi.repository import GObject
  5. from lib.commands import ControlServerCommands
  6. from lib.tcpmulticonnection import TCPMultiConnection
  7. from lib.response import NotifyResponse, OkResponse
  8. class ControlServer(TCPMultiConnection):
  9. def __init__(self, pipeline):
  10. '''Initialize server and start listening.'''
  11. self.log = logging.getLogger('ControlServer')
  12. super().__init__(port=9999)
  13. self.command_queue = Queue()
  14. self.commands = ControlServerCommands(pipeline)
  15. def on_accepted(self, conn, addr):
  16. '''Asynchronous connection listener. Starts a handler for each connection.'''
  17. self.log.debug('setting gobject io-watch on connection')
  18. GObject.io_add_watch(conn, GObject.IO_IN, self.on_data, [''])
  19. def on_data(self, conn, _, leftovers, *args):
  20. '''Asynchronous connection handler. Pushes data from socket
  21. into command queue linewise'''
  22. close_after = False
  23. try:
  24. while True:
  25. try:
  26. leftovers.append(conn.recv(4096).decode(errors='replace'))
  27. if len(leftovers[-1]) == 0:
  28. self.log.info("Socket was closed")
  29. leftovers.pop()
  30. close_after = True
  31. break
  32. except UnicodeDecodeError as e:
  33. continue
  34. except:
  35. pass
  36. data = "".join(leftovers)
  37. del leftovers[:]
  38. lines = data.split('\n')
  39. for line in lines[:-1]:
  40. self.log.debug("got line: %r", line)
  41. line = line.strip()
  42. # TODO: move quit to on_loop
  43. # 'quit' = remote wants us to close the connection
  44. if line == 'quit':
  45. self.log.info("Client asked us to close the Connection")
  46. self.close_connection(conn)
  47. return False
  48. if self.command_queue.empty():
  49. self.log.debug('command_queue was empty, re-starting on_loop scheduling')
  50. GObject.idle_add(self.on_loop)
  51. self.command_queue.put((line, conn))
  52. if close_after:
  53. self.close_connection(conn)
  54. return False
  55. if lines[-1] != '':
  56. self.log.debug("remaining %r", lines[-1])
  57. leftovers.append(lines[-1])
  58. return True
  59. def on_loop(self):
  60. '''Command handler. Processes commands in the command queue whenever
  61. nothing else is happening (registered as GObject idle callback)'''
  62. self.log.debug('on_loop called')
  63. if self.command_queue.empty():
  64. self.log.debug('command_queue is empty again, stopping on_loop scheduling')
  65. return False
  66. line, requestor = self.command_queue.get()
  67. words = line.split()
  68. if len(words) < 1:
  69. self.log.debug('command_queue is empty again, stopping on_loop scheduling')
  70. return False
  71. command = words[0]
  72. args = words[1:]
  73. self.log.info("processing command %r with args %s", command, args)
  74. response = None
  75. try:
  76. # deny calling private methods
  77. if command[0] == '_':
  78. self.log.info('private methods are not callable')
  79. raise KeyError()
  80. command_function = self.commands.__class__.__dict__[command]
  81. except KeyError as e:
  82. self.log.info("received unknown command %s", command)
  83. response = "error unknown command %s\n" % command
  84. else:
  85. try:
  86. responseObject = command_function(self.commands, *args)
  87. except Exception as e:
  88. message = str(e) or "<no message>"
  89. response = "error %s\n" % message
  90. else:
  91. if isinstance(responseObject, NotifyResponse):
  92. responseObject = [ responseObject ]
  93. if isinstance(responseObject, list):
  94. for obj in responseObject:
  95. signal = "%s\n" % str(obj)
  96. for conn in self.currentConnections:
  97. self._schedule_write(conn, signal)
  98. else:
  99. response = "%s\n" % str(responseObject)
  100. finally:
  101. if response is not None and requestor in self.currentConnections:
  102. self._schedule_write(requestor, response)
  103. return False
  104. def _schedule_write(self, conn, message):
  105. queue = self.currentConnections[conn]
  106. if queue.empty():
  107. self.log.debug('write_queue[%u] was empty, re-starting on_write scheduling', conn.fileno())
  108. GObject.io_add_watch(conn, GObject.IO_OUT, self.on_write)
  109. queue.put(message)
  110. def on_write(self, conn, *args):
  111. # TODO: on_loop() is not called as soon as there is a writable socket
  112. self.on_loop()
  113. self.log.debug('on_write[%u] called', conn.fileno())
  114. try:
  115. queue = self.currentConnections[conn]
  116. except KeyError:
  117. return False
  118. if queue.empty():
  119. self.log.debug('write_queue[%u] is empty again, stopping on_write scheduling', conn.fileno())
  120. return False
  121. message = queue.get()
  122. try:
  123. conn.send(message.encode())
  124. except Exception as e:
  125. self.log.warn(e)
  126. return True
  127. def notify_all(self, msg):
  128. try:
  129. words = msg.split()
  130. words[-1] = self.commands.encodeSourceName(int(words[-1]))
  131. msg = " ".join(words) + '\n'
  132. for conn in self.currentConnections:
  133. self._schedule_write(conn, msg)
  134. except Exception as e:
  135. self.log.debug("error during notify: %s", e)