aboutsummaryrefslogtreecommitdiff
path: root/voctocore/lib/controlserver.py
blob: 22dd30539508286a8fb85c60dce1c245481313bf (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. GObject.idle_add(self.on_loop)
  16. def on_accepted(self, conn, addr):
  17. '''Asynchronous connection listener. Starts a handler for each connection.'''
  18. self.log.debug('setting gobject io-watch on connection')
  19. GObject.io_add_watch(conn, GObject.IO_IN, self.on_data, [''])
  20. GObject.io_add_watch(conn, GObject.IO_OUT, self.on_write)
  21. def on_data(self, conn, _, leftovers, *args):
  22. '''Asynchronous connection handler. Pushes data from socket
  23. into command queue linewise'''
  24. close_after = False
  25. try:
  26. while True:
  27. try:
  28. leftovers.append(conn.recv(4096).decode(errors='replace'))
  29. if len(leftovers[-1]) == 0:
  30. self.log.info("Socket was closed")
  31. leftovers.pop()
  32. close_after = True
  33. break
  34. except UnicodeDecodeError as e:
  35. continue
  36. except BlockingIOError as e:
  37. pass
  38. data = "".join(leftovers)
  39. leftovers.clear()
  40. lines = data.split('\n')
  41. for line in lines[:-1]:
  42. self.log.debug("got line: %r", line)
  43. line = line.strip()
  44. # TODO: move quit to on_loop
  45. # 'quit' = remote wants us to close the connection
  46. if line == 'quit':
  47. self.log.info("Client asked us to close the Connection")
  48. self.close_connection(conn)
  49. return False
  50. self.command_queue.put((line, conn))
  51. if close_after:
  52. self.close_connection(conn)
  53. return False
  54. if lines[-1] != '':
  55. self.log.debug("remaining %r", lines[-1])
  56. leftovers.append(lines[-1])
  57. return True
  58. def on_loop(self):
  59. '''Command handler. Processes commands in the command queue whenever
  60. nothing else is happening (registered as GObject idle callback)'''
  61. if self.command_queue.empty():
  62. return True
  63. line, requestor = self.command_queue.get()
  64. words = line.split()
  65. if len(words) < 1:
  66. return True
  67. command = words[0]
  68. args = words[1:]
  69. self.log.info("processing command %r with args %s", command, args)
  70. try:
  71. command_function = self.commands.__class__.__dict__[command]
  72. except KeyError as e:
  73. self.log.info("received unknown command %s", command)
  74. response = "error unknown command %s\n" % command
  75. else:
  76. try:
  77. responseObject = command_function(self.commands, *args)
  78. except Exception as e:
  79. message = str(e) or "<no message>"
  80. response = "error %s\n" % message
  81. else:
  82. if isinstance(responseObject, NotifyResponse):
  83. signal = "%s\n" % str(responseObject)
  84. for conn, queue in self.currentConnections.items():
  85. if conn == requestor:
  86. continue
  87. queue.put(signal)
  88. response = "%s\n" % str(responseObject)
  89. finally:
  90. if requestor in self.currentConnections:
  91. self.currentConnections[requestor].put(response)
  92. return True
  93. def on_write(self, conn, *args):
  94. # TODO: on_loop() is not called as soon as there is a writable socket
  95. self.on_loop()
  96. try:
  97. queue = self.currentConnections[conn]
  98. except KeyError:
  99. return False
  100. if queue.empty():
  101. return True
  102. message = queue.get()
  103. try:
  104. conn.send(message.encode())
  105. except Exception as e:
  106. self.log.warn(e)
  107. return True
  108. def notify_all(self, msg):
  109. try:
  110. words = msg.split()
  111. words[-1] = self.commands.encodeSourceName(int(words[-1]))
  112. msg = " ".join(words) + '\n'
  113. for queue in self.currentConnections.values():
  114. queue.put(msg)
  115. except Exception as e:
  116. self.log.debug("error during notify: %s", e)