flask-timing/job.py at master · limboinf/flask-timing · GitHub

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
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
# coding=utf-8
"""
beanstald job
    :copyright: (c) 2015 by fangpeng.
    :license: MIT, see LICENSE for more details.
"""
import time
import json
import logging
from beanstalkc import Connection


class JobService(object):
    """
    beanstalkd job service
    watching the appointed tube.
    """
    def __init__(self, host, port, tube):
        self.__host = host
        self.__port = port
        self.__tube = tube

    def connect(self):
        while True:
            try:
                self.__bs = Connection(self.__host, self.__port)
                print 'connected to %s:%s' % (self.__host, self.__port)
                self.__bs.watch(self.__tube)
                print 'watching %s' % self.__bs.watching()
                if not self.__bs is None:
                    return
            except:
                time.sleep(1)

    def process(self, job):
        """overwrite
        Realize your functions
        """
        pass

    def finish(self, job):
        job.delete()

    def start(self):
        self.connect()
        while True:
            try:
                job = self.__bs.reserve()
                print '>>[new job]', job, job.body
                self.process(job)
                self.finish(job)
            except Exception as ex:
                logging.error('job process exception', exc_info=ex)
                self.connect()


class JobWorkerClient(object):
    """
    job client.
    """
    def __init__(self, host, port, tube=None):
        self.__host = host
        self.__port = port
        self.__tube = tube
        self.__bs = None

    def _ensure_connect(self):
        if self.__bs is None:
            print 'beanstalkc client connecting ...'
            self.__bs = Connection(self.__host, self.__port)
            print 'beanstalkc client connected to %s:%s:[%s]' % (self.__host, self.__port, self.__tube)
            if not self.__tube is None:
                self.__bs.use(self.__tube)
                print 'beanstalkc client using %s' % self.__bs.using()

    def put(self,obj):
        self._ensure_connect()
        self.__bs.put(json.dumps(obj))

    def use_put(self, tube, obj):
        self._ensure_connect()
        self.__bs.use(tube)
        self.__bs.put(json.dumps(obj))

    def use(self, tube):
        self._ensure_connect()
        self.__bs.use(tube)
        self.__tube = tube

    def close(self):
        try:
            self.__bs.close()
        except:
            pass