Mercurial > ~darius > hgwebdir.cgi > epro

comparison velib_python/streamcommand.py @ 8:9c0435a617db

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Import velib_python
author Daniel O'Connor <darius@dons.net.au>
date Sun, 05 Dec 2021 14:35:36 +1030
parents
children
comparison
equal deleted inserted replaced
5:982eeffe9d95 8:9c0435a617db
1 #!/usr/bin/env python3
2 # -*- coding: utf-8 -*-
3
4 ## @package dbus_vrm
5
6 import logging
7 logger = logging.getLogger(__name__)
8
9 import codecs
10 import threading
11 import subprocess
12 from time import sleep
13
14 # Runs a command, and calls sendfeedback with the statusupdates.
15 class StreamCommand(object):
16 SIGNALS = {
17 1: "SIGHUP", 2: "SIGINT", 3: "SIGQUIT", 4: "SIGILL", 6: "SIGABRT", 7: "SIGBUS", 8: "SIGFPE",
18 9: "SIGKILL", 10: "SIGBUS", 11: "SIGSEGV", 12: "SIGSYS", 13: "SIGPIPE", 14: "SIGALRM",
19 15: "SIGTERM"}
20
21 def run(self, command, timeout, feedbacksender):
22 self.feedbacksender = feedbacksender
23 self.returncode = None
24 self.utf8_decoder = codecs.getdecoder("utf_8")
25 self.latin1_decoder = codecs.getdecoder("latin1")
26
27 def target():
28 logger.info('Thread started for running %s' % command)
29 self.feedbacksender.send({"status": "starting"})
30
31 try:
32 self.process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
33 except OSError as e:
34 logger.info("Command %s could not be started, errno: %s, msg: %s"
35 % (command, e.errno, e.strerror))
36 self.feedbacksender.send({"status": "error",
37 "errormessage": "Could not start (errno %s, msg %s)" % (e.errno, e.strerror),
38 "errorcode": 731}, finished=True)
39
40 self.process = None
41 return
42
43 self.readandsend()
44
45
46 thread = threading.Thread(target=target)
47 thread.start()
48 thread.join(timeout)
49
50 if self.process is None:
51 # Error message has already beent sent
52 return None
53
54 # Make sure to send all the output
55 self.readandsend()
56
57 if thread.is_alive():
58 logger.warning("Command %s will now be terminated because of timeout" % command)
59 self.process.terminate() # TODO or should it be killed?
60 thread.join()
61 logger.warning("Command %s has been terminated" % command)
62 r = {"status": "error", "errormessage": "Stopped by timeout", "errorcode": 732}
63
64 elif self.process.returncode < 0:
65 signal = -1 * self.process.returncode
66 error = "Stopped with signal %d - %s" % (signal, self.SIGNALS.get(signal, "unknown"))
67 logger.warning("Command %s abnormal stop. %s" % (command, error))
68 r = {"status": "error", "errorcode": 733, "errormessage": error}
69
70 else:
71 logger.info("Command %s execution completed. Exitcode %d" % (command, self.process.returncode))
72 r = {"status": "finished", "exitcode": self.process.returncode}
73
74 self.feedbacksender.send(r, finished=True)
75 return self.process.returncode
76
77 def readandsend(self):
78 # TODO: check that below code works OK with vup stdout encoding (UTF-8), including non-standard ASCII chars
79
80 while True:
81 self.process.stdout.flush()
82 line = self.process.stdout.readline()
83 try:
84 unicode_line, _ = self.utf8_decoder(line)
85 except UnicodeDecodeError:
86 unicode_line, _ = self.latin1_decoder(line)
87
88 # Max length on pubnub is 1800 chars, and output is much better readable with the bare eye
89 # when sent per line. So no need to send it alltogether.
90 self.feedbacksender.send({"status": "running", "xmloutput": unicode_line})
91 if line == b'' and self.process.poll() != None:
92 break
93 sleep(0.04)