9 from ..model.job import exJobs |
9 from ..model.job import exJobs |
10 |
10 |
11 from .pool import taskPool |
11 from .pool import taskPool |
12 |
12 |
13 class Task: |
13 class Task: |
14 '''one single part of a job. |
14 '''A Task is one message to one recipient and is a part of one :class:`iro.model.job.ExJob`. |
15 a task is one message to one recipient |
|
16 ''' |
15 ''' |
17 def __init__(self, recipient, job): |
16 def __init__(self, recipient, job): |
|
17 """ |
|
18 :param recipient: a recipient |
|
19 :param `iro.model.job.ExJob` job: connected job |
|
20 """ |
18 self.recipient = recipient |
21 self.recipient = recipient |
19 self.job = job |
22 self.job = job |
20 self.status = None |
23 self.status = None |
21 self.error = False |
24 self.error = False |
22 |
25 |
23 def setStatus(self,status): |
26 def setStatus(self,status): |
|
27 """callback, to set status of task""" |
24 self.status = status |
28 self.status = status |
25 return status |
29 return status |
26 |
30 |
27 def setError(self, error): |
31 def setError(self, error): |
|
32 """errback to set error of task""" |
28 self.status = error |
33 self.status = error |
29 self.error = True |
34 self.error = True |
30 return error |
35 return error |
31 |
36 |
32 def start(self): |
37 def start(self): |
|
38 """Starting to send message to recipient. |
|
39 |
|
40 :return: a defer, that is fired, when message is sended successfully over any offer. |
|
41 """ |
33 self.d = Deferred() |
42 self.d = Deferred() |
34 self.d.addCallback(self.setStatus) |
43 self.d.addCallback(self.setStatus) |
35 self.d.addCallback(partial(self.job.setStatus,self)) |
44 self.d.addCallback(partial(self.job.setStatus,self)) |
36 self.d.addErrback(self.setError) |
45 self.d.addErrback(self.setError) |
37 self.d.addErrback(partial(self.job.setError,self)) |
46 self.d.addErrback(partial(self.job.setError,self)) |
38 taskPool.run(self._run) |
47 taskPool.run(self._run) |
39 return self.d |
48 return self.d |
40 |
49 |
41 def _run(self): |
50 def _run(self): |
|
51 """sends the message to recipient, tries all possible offers.""" |
42 os= iter(self.job.offers) |
52 os= iter(self.job.offers) |
43 def n(): |
53 def n(): |
44 try: |
54 try: |
45 offer = os.next() |
55 offer = os.next() |
46 d = maybeDeferred(offers[offer],self.recipient,self.job.message) |
56 d = maybeDeferred(offers[offer],self.recipient,self.job.message) |
59 n() |
69 n() |
60 |
70 |
61 |
71 |
62 @inlineCallbacks |
72 @inlineCallbacks |
63 def createJob(user,recipients, msg, offers, info=None): |
73 def createJob(user,recipients, msg, offers, info=None): |
|
74 """Creates a :class:`iro.model.job.ExJob` and start for all recipients one task. |
|
75 |
|
76 :param `iro.model.schema.User` user: the sender |
|
77 :param `iro.model.message.Message` msg: the message |
|
78 :param list offers: a list of possible offer and provider names, to try to send the message over. The first entry will be tried first. |
|
79 :param string info: a bill group name |
|
80 :return: the new :class:`iro.model.job.ExJob` object. |
|
81 """ |
64 job = yield exJobs.create(user, recipients, msg, offers, info) |
82 job = yield exJobs.create(user, recipients, msg, offers, info) |
65 for r in recipients: |
83 for r in recipients: |
66 task = Task(r,job) |
84 task = Task(r,job) |
67 job.addTask(task) |
85 job.addTask(task) |
68 task.start() |
86 task.start() |