diff -rc Mixminion-0.0.8alpha3/lib/mixminion/server/ServerQueue.py Mixminion-0.0.8alpha3.patched/lib/mixminion/server/ServerQueue.py *** Mixminion-0.0.8alpha3/lib/mixminion/server/ServerQueue.py 2007-09-12 21:42:48.000000000 +0200 --- Mixminion-0.0.8alpha3.patched/lib/mixminion/server/ServerQueue.py 2010-11-05 12:00:04.000000000 +0100 *************** *** 617,632 **** try: for ds in self.store._metadata_cache.values(): if not self.addressStateDB.has_key(str(ds.address)): ! as = _AddressState(ds.address) ! self.addressStateDB[str(ds.address)] = as if not self.retrySchedule: rs = [3600] self.totalLifetime = 3600 else: rs = self.retrySchedule self.totalLifetime = reduce(operator.add,self.retrySchedule,0) ! for as in self.addressStateDB.values(): ! as.setNextAttempt(rs, now) self._repOK() finally: self._lock.release() --- 617,632 ---- try: for ds in self.store._metadata_cache.values(): if not self.addressStateDB.has_key(str(ds.address)): ! ass = _AddressState(ds.address) ! self.addressStateDB[str(ds.address)] = ass if not self.retrySchedule: rs = [3600] self.totalLifetime = 3600 else: rs = self.retrySchedule self.totalLifetime = reduce(operator.add,self.retrySchedule,0) ! for ass in self.addressStateDB.values(): ! ass.setNextAttempt(rs, now) self._repOK() finally: self._lock.release() *************** *** 643,652 **** else: have[ds.address]=1 ! for k, as in self.addressStateDB.items(): ! if have.has_key(as.address): continue ! lastActivity = as.getLastActivity() if lastActivity and ( lastActivity + self.totalLifetime < now): del self.addressStateDB[k] --- 643,652 ---- else: have[ds.address]=1 ! for k, ass in self.addressStateDB.items(): ! if have.has_key(ass.address): continue ! lastActivity = ass.getLastActivity() if lastActivity and ( lastActivity + self.totalLifetime < now): del self.addressStateDB[k] *************** *** 655,665 **** def _getAddressState(self, address, now=None): try: ! as = self.addressStateDB[str(address)] except KeyError: ! as = self.addressStateDB[str(address)] = _AddressState(address) ! as.setNextAttempt(self.retrySchedule, now) ! return as def queueDeliveryMessage(self, msg, address, now=None): self._getAddressState(address, now=now) --- 655,665 ---- def _getAddressState(self, address, now=None): try: ! ass = self.addressStateDB[str(address)] except KeyError: ! ass = self.addressStateDB[str(address)] = _AddressState(address) ! ass.setNextAttempt(self.retrySchedule, now) ! return ass def queueDeliveryMessage(self, msg, address, now=None): self._getAddressState(address, now=now) *************** *** 767,774 **** self._repOK() o = self.store.getObject(handle) ds = self.store.getMetadata(handle) ! as = self._getAddressState(ds.address) ! return (o, ds, as) def _repOK(self): """Raise an assertion error if the internal state of this object is --- 767,774 ---- self._repOK() o = self.store.getObject(handle) ds = self.store.getMetadata(handle) ! ass = self._getAddressState(ds.address) ! return (o, ds, ass) def _repOK(self): """Raise an assertion error if the internal state of this object is *************** *** 780,787 **** DeliveryQueue._repOK(self) for h in self.store._metadata_cache.keys(): ds = self.store._metadata_cache[h] ! as = self._getAddressState(ds.address) ! assert as.address == ds.address finally: self._lock.release() --- 780,787 ---- DeliveryQueue._repOK(self) for h in self.store._metadata_cache.keys(): ds = self.store._metadata_cache[h] ! ass = self._getAddressState(ds.address) ! assert ass.address == ds.address finally: self._lock.release() diff -rc Mixminion-0.0.8alpha3/lib/mixminion/test.py Mixminion-0.0.8alpha3.patched/lib/mixminion/test.py *** Mixminion-0.0.8alpha3/lib/mixminion/test.py 2005-12-02 20:56:35.000000000 +0100 --- Mixminion-0.0.8alpha3.patched/lib/mixminion/test.py 2010-11-05 11:59:37.000000000 +0100 *************** *** 3444,3458 **** h3 = q.queueDeliveryMessage("Message number three", A1, start) q._repOK() ! m, ds, as = q._inspect(h1) self.assertEquals(m, "Message number one") self.assertEquals(ds.address, A1) ! self.assertEquals(as.address, A1) self.assert_(not ds.isPending()) self.assertEquals(ds.lastAttempt, None) ! self.assertEquals(as.lastSuccess, None) ! self.assertEquals(as.lastFailure, None) ! self.assertEquals(as.firstFailure, None) q.removeExpiredMessages(start) self.assertEquals(q.store.count(), 3) --- 3444,3458 ---- h3 = q.queueDeliveryMessage("Message number three", A1, start) q._repOK() ! m, ds, ass = q._inspect(h1) self.assertEquals(m, "Message number one") self.assertEquals(ds.address, A1) ! self.assertEquals(ass.address, A1) self.assert_(not ds.isPending()) self.assertEquals(ds.lastAttempt, None) ! self.assertEquals(ass.lastSuccess, None) ! self.assertEquals(ass.lastFailure, None) ! self.assertEquals(ass.firstFailure, None) q.removeExpiredMessages(start) self.assertEquals(q.store.count(), 3) *************** *** 3470,3485 **** msgs[h3].failed(now=start+40, retriable=1) self.assert_(not q._inspect(h3)[1].isPending()) ! as = q.addressStateDB[str(A1)] ! self.assertIEquals(as.lastSuccess-start, 30) ! self.assertIEquals(as.firstFailure-start, 10) ! self.assertIEquals(as.lastFailure-start, 10) ! self.assertIEquals(as.nextAttempt-start, 10+HOUR) ! as = q.addressStateDB[str(A2)] ! self.assertIEquals(as.lastSuccess-start, 20) ! self.assertEquals(as.firstFailure, None) ! self.assertEquals(as.lastFailure, None) ! self.assertIEquals(as.nextAttempt-start, 20) # Queue two more messages; the one on A2 gets tried; the ones on # A1 doesn't get tried for a while. --- 3470,3485 ---- msgs[h3].failed(now=start+40, retriable=1) self.assert_(not q._inspect(h3)[1].isPending()) ! ass = q.addressStateDB[str(A1)] ! self.assertIEquals(ass.lastSuccess-start, 30) ! self.assertIEquals(ass.firstFailure-start, 10) ! self.assertIEquals(ass.lastFailure-start, 10) ! self.assertIEquals(ass.nextAttempt-start, 10+HOUR) ! ass = q.addressStateDB[str(A2)] ! self.assertIEquals(ass.lastSuccess-start, 20) ! self.assertEquals(ass.firstFailure, None) ! self.assertEquals(ass.lastFailure, None) ! self.assertIEquals(ass.nextAttempt-start, 20) # Queue two more messages; the one on A2 gets tried; the ones on # A1 doesn't get tried for a while. *************** *** 3495,3507 **** q.sendReadyMessages(start+HOUR+MIN) msgs = self._pendingMsgDict(q._msgs) self.assertUnorderedEq(msgs.keys(), [h3, h4]) ! as = q.addressStateDB[str(A1)] msgs[h3].failed(now=start+HOUR+MIN+5, retriable=1) msgs[h4].failed(now=start+HOUR+MIN+5, retriable=1) ! self.assertIEquals(as.firstFailure-start, 10) ! self.assertIEquals(as.lastFailure-start, HOUR+MIN) ! self.assertIEquals(as.nextAttempt-start, 10+2*HOUR) h6 = q.queueDeliveryMessage("The sixth message", A1, now=start+HOUR*17-MIN) --- 3495,3507 ---- q.sendReadyMessages(start+HOUR+MIN) msgs = self._pendingMsgDict(q._msgs) self.assertUnorderedEq(msgs.keys(), [h3, h4]) ! ass = q.addressStateDB[str(A1)] msgs[h3].failed(now=start+HOUR+MIN+5, retriable=1) msgs[h4].failed(now=start+HOUR+MIN+5, retriable=1) ! self.assertIEquals(ass.firstFailure-start, 10) ! self.assertIEquals(ass.lastFailure-start, HOUR+MIN) ! self.assertIEquals(ass.nextAttempt-start, 10+2*HOUR) h6 = q.queueDeliveryMessage("The sixth message", A1, now=start+HOUR*17-MIN) *************** *** 3525,3537 **** self.assertUnorderedEq(q.addressStateDB.keys(), [str(A1)]) # Make sure that messages keep getting retried... ! as = q.addressStateDB[str(A1)] # (Reset nextDelivery on A1, since not all the delivery attempts # have really happened.) q.sendReadyMessages(now=start+HOUR*17+MIN*15) msgs = self._pendingMsgDict(q._msgs) msgs[h6].failed(now=start+HOUR*17+MIN*16, retriable=0) ! self.assertEquals(as.nextAttempt - start, 25*HOUR+10) self.assertEquals([], q.store.getAllMessages()) # Test reloading. --- 3525,3537 ---- self.assertUnorderedEq(q.addressStateDB.keys(), [str(A1)]) # Make sure that messages keep getting retried... ! ass = q.addressStateDB[str(A1)] # (Reset nextDelivery on A1, since not all the delivery attempts # have really happened.) q.sendReadyMessages(now=start+HOUR*17+MIN*15) msgs = self._pendingMsgDict(q._msgs) msgs[h6].failed(now=start+HOUR*17+MIN*16, retriable=0) ! self.assertEquals(ass.nextAttempt - start, 25*HOUR+10) self.assertEquals([], q.store.getAllMessages()) # Test reloading. diff -rc Mixminion-0.0.8alpha3/lib/mixminion/ThreadUtils.py Mixminion-0.0.8alpha3.patched/lib/mixminion/ThreadUtils.py *** Mixminion-0.0.8alpha3/lib/mixminion/ThreadUtils.py 2005-06-04 15:54:14.000000000 +0200 --- Mixminion-0.0.8alpha3.patched/lib/mixminion/ThreadUtils.py 2010-11-05 11:59:53.000000000 +0100 *************** *** 43,48 **** --- 43,53 ---- class ClearableQueue(MessageQueue): """Extended version of python's Queue class that supports removing all the items from the queue.""" + + # Check whether the queue is empty + def _empty(self): + return not self.queue + def clear(self): """Remove all the items from this queue.""" # If the queue is empty, return.