PyQt Flow Signals

Question

I deal with PyQt and stream signals / slots. Here is a situation where I cannot find my error:

I have a class (MultipleProcessLauncher) that is able to run multiple processes in separate threads. I catch the stdout of each process and send these messages in a single queue that is read by another thread (OutputWorker), this last thread should send an onNewMessage signal (I think it is not) to the main class, but the callback function is never called.

• Process threads fill the queue with messages
• The read stream captures all of these messages (I can print them using print(item)in a while loop)

But : - The read thread signal does not seem to emit anything, so the main thread callback function is never called ...

Your help would be much appreciated, I think I'm missing something with cross-flow signals ...

class OutputWorker(QObject):
    onNewMessage = pyqtSignal(['QString'])

    def __init__(self, queue, parent=None):
        super(OutputWorker, self).__init__(parent)
        self.queue = queue

    def work(self):
        while True:
            item = self.queue.get()
            self.onNewMessage.emit(item)
            self.queue.task_done()

class MultipleProcessLauncher(QObject):
    commandEvent = pyqtSignal(['QString'])

    def __init__(self, parent=None):
        super(MultipleProcessLauncher, self).__init__(parent)

        self.messaging_queue = Queue()

        # Start reading message
        self.reading_thread = QThread()

        self.worker = OutputWorker(self.messaging_queue)
        self.worker.moveToThread(self.reading_thread)
        self.worker.onNewMessage.connect(self.command_event)

        self.reading_thread.started.connect(self.worker.work)
        self.reading_thread.start()

    def execute(self, command):
        p = subprocess.Popen(command, stdout=subprocess.PIPE)
        t = Thread(target=self.enqueue, args=(p.stdout, self.messaging_queue))
        t.daemon = True
        t.start()

    def enqueue(self, stdout, queue):
        for line in iter(stdout.readline, b''):
            queue.put(line.decode())
        stdout.close()

    def command_event(self, event):
        # This point is never reached
        print('message received')

if __name__ == '__main__':
    manager = MultipleProcessLauncher()
    manager.execute('ipconfig')

    time.sleep(100)

Answer 1

Qt cross-threading is based on an event loop, so you need to do QApplication so that there is a main event loop to process signals from other threads. For instance:

if __name__ == '__main__':
    app = QApplication([])
    manager = MultipleProcessLauncher()
    manager.execute('ipconfig')
    MAX_WAIT_MSEC = 100 * 1000  # 100 seconds
    QTimer.singleShot(MAX_WAIT_MSEC, app.quit) 
    app.exec()

In your real application, you will probably execute the manager based on user input, so the execution will be in the slot, and there would be no need to stop working, etc., but you will get this idea.