| 39 | |
| 40 | |
| 41 | class MBus: |
| 42 | |
| 43 | def __init__ (self, db, name): |
| 44 | self.db = db |
| 45 | self.name = name |
| 46 | self.seen = self._find_last_id() |
| 47 | self.mbox = [] |
| 48 | |
| 49 | # PRIVATE |
| 50 | def _find_last_id (self): |
| 51 | return self.db.execute(FIND_LAST_ID, ()).fetchone()[0] or 1 |
| 52 | |
| 53 | def _poll (self): |
| 54 | """Fetch new messages from database and append to mailbox. |
| 55 | """ |
| 56 | for row in list(self.db.execute(RECV_MESSAGES, (self.seen,))): |
| 57 | self.seen, source, dest, blob = row |
| 58 | if source != self.name and fnmatch.fnmatch(self.name, dest): |
| 59 | tag, data = cPickle.loads(str(blob)) |
| 60 | self.mbox.append((self.seen, source, tag, data)) |
| 61 | |
| 62 | def _filter (self, tag, func): |
| 63 | """Remove and return matching messages from mailbox and retain the rest. |
| 64 | """ |
| 65 | mbox = [] |
| 66 | for t in self.mbox: |
| 67 | if fnmatch.fnmatch(t[2], tag) and func(t): |
| 68 | yield t |
| 69 | else: |
| 70 | mbox.append(t) |
| 71 | self.mbox = mbox |
| 72 | |
| 73 | # PUBLIC |
| 74 | def recv (self, tag='*', func=lambda _: True, wait=5, sleep=0.5): |
| 75 | end = time.time() + wait |
| 76 | while True: |
| 77 | self._poll() |
| 78 | for t in self._filter(tag, func): |
| 79 | yield t |
| 80 | if time.time() > end: |
| 81 | break |
| 82 | time.sleep(sleep) |
| 83 | |
| 84 | def send (self, dest, tag, **kwargs): |
| 85 | data = (tag, kwargs) |
| 86 | rowid = self.db.execute(SEND_MESSAGE, (self.name, dest, |
| 87 | cPickle.dumps(data))).lastrowid |
| 88 | return rowid |
| 89 | |
| 90 | |
| 91 |
no outgoing calls
no test coverage detected