devel.cz

# -*- coding: utf-8 -*-
from multiprocessing import Process
import signal
from threading import Lock
 
 
class Task(Process):
 
    def __init__(self, data):
        super(Task, self).__init__()
        self.data = data
 
    def run(self):
        signal.signal(signal.SIGCHLD, signal.SIG_DFL)
        self.do_job() # long job there
        pass
 
    def do_job(self):
        # very long job
        pass
 
 
class MQListener(object):
 
    def __init__(self):
        self.tasks = []
        self.tasks_lock = Lock()
        self.register_signal_handler()
        mq = RabbitMQ()
        mq.listen("task_queue", self.on_message)
 
    def register_signal_handler(self):
        signal.signal(signal.SIGCHLD, self.on_signal_received)
 
    def on_signal_received(self, *_):
        self._check_existing_processes()
 
    def on_message(self, message):
        # ack message and create task
        task = Task(message)
        with self.tasks_lock:
            self.tasks.append(task)
            task.start()
        pass
 
    def _check_existing_processes(self):
        """
        go over all created task, if some is not alive - remove them from tasks collection
        """
        try:
            with self.tasks_lock:
                running_tasks = []
                for w in self.tasks:
                    if not w.is_alive():
                        w.join()
                    else:
                        running_tasks.append(w)
 
                self.tasks = running_tasks
        except Exception:
            # log
            pass
 
 
 
if __name__ == '__main__':
    m = MQListener()