File: tasksconsumer.py

package info (click to toggle)
uwsgi 2.0.31-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 6,564 kB
  • sloc: ansic: 87,066; python: 7,004; cpp: 1,133; java: 708; perl: 678; sh: 585; ruby: 555; makefile: 148; xml: 130; cs: 121; objc: 37; php: 28; erlang: 20; javascript: 11
file content (46 lines) | stat: -rw-r--r-- 1,064 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
from uwsgidecorators import spool
from threading import Thread

from six.moves import queue

queues = {}


class queueconsumer(object):

    def __init__(self, name, num=1, **kwargs):
        self.name = name
        self.num = num
        self.queue = queue.Queue()
        self.threads = []
        self.func = None
        queues[self.name] = self

    @staticmethod
    def consumer(self):
        while True:
            req = self.queue.get()
            print(req)
            self.func(req)
            self.queue.task_done()

    def __call__(self, f):
        self.func = f
        for i in range(self.num):
            t = Thread(target=self.consumer, args=(self,))
            self.threads.append(t)
            t.daemon = True
            t.start()


@spool
def spooler_enqueuer(arguments):
    if 'queue' in arguments:
        queue = arguments['queue']
        queues[queue].queue.put(arguments)
    else:
        raise Exception("You have to specify a queue name")


def enqueue(*args, **kwargs):
    return spooler_enqueuer.spool(*args, **kwargs)