[CalendarServer-changes] [14786] twext/trunk/twext/enterprise

[source_changes at macosforge.org]

Revision: 14786
          http://trac.calendarserver.org//changeset/14786
Author:   cdaboo at apple.com
Date:     2015-05-14 11:17:39 -0700 (Thu, 14 May 2015)
Log Message:
-----------
Add option to enable/disable the job queue processing.

Modified Paths:
--------------
    twext/trunk/twext/enterprise/jobqueue.py
    twext/trunk/twext/enterprise/test/test_jobqueue.py

Modified: twext/trunk/twext/enterprise/jobqueue.py
===================================================================
--- twext/trunk/twext/enterprise/jobqueue.py	2015-05-14 15:45:58 UTC (rev 14785)
+++ twext/trunk/twext/enterprise/jobqueue.py	2015-05-14 18:17:39 UTC (rev 14786)
@@ -1952,7 +1952,7 @@
     highPriorityLevel = 80      # Percentage load level above which only high priority jobs are processed
     mediumPriorityLevel = 50    # Percentage load level above which high and medium priority jobs are processed
 
-    def __init__(self, reactor, transactionFactory, ampPort, useWorkerPool=True):
+    def __init__(self, reactor, transactionFactory, ampPort, useWorkerPool=True, disableWorkProcessing=False):
         """
         Initialize a L{PeerConnectionPool}.
 
@@ -1977,6 +1977,7 @@
         self.ampPort = ampPort
         self.thisProcess = None
         self.workerPool = WorkerConnectionPool() if useWorkerPool else None
+        self.disableWorkProcessing = disableWorkProcessing
         self.peers = []
         self.mappedPeers = {}
         self._startingUp = None
@@ -1995,6 +1996,20 @@
         self.peers.append(peer)
 
 
+    def enable(self):
+        """
+        Turn on work queue processing.
+        """
+        self.disableWorkProcessing = False
+
+
+    def disable(self):
+        """
+        Turn off work queue processing.
+        """
+        self.disableWorkProcessing = True
+
+
     def totalLoad(self):
         return self.workerPool.allWorkerLoad() if self.workerPool else 0
 
@@ -2096,7 +2111,7 @@
 
         loopCounter = 0
         while True:
-            if not self.running:
+            if not self.running or self.disableWorkProcessing:
                 returnValue(None)
 
             # Check the overall service load - if overloaded skip this poll cycle.

Modified: twext/trunk/twext/enterprise/test/test_jobqueue.py
===================================================================
--- twext/trunk/twext/enterprise/test/test_jobqueue.py	2015-05-14 15:45:58 UTC (rev 14785)
+++ twext/trunk/twext/enterprise/test/test_jobqueue.py	2015-05-14 18:17:39 UTC (rev 14786)
@@ -1269,7 +1269,44 @@
         self.assertEqual(jobs[1].notBefore, fakeNow - datetime.timedelta(20 * 60, 5))
 
 
+    @inlineCallbacks
+    def test_enableDisable(self):
+        """
+        L{PeerConnectionPool.enable} and L{PeerConnectionPool.disable} control queue processing.
+        """
+        dbpool, qpool, clock, performerChosen = self._setupPools()
 
+        # Disable processing
+        qpool.disable()
+
+        @transactionally(dbpool.pool.connection)
+        def check(txn):
+            return qpool.enqueueWork(
+                txn, DummyWorkItem, a=3, b=9,
+                notBefore=datetime.datetime(2012, 12, 12, 12, 12, 0)
+            )
+
+        yield check
+
+        # Advance far beyond the given timestamp.
+        clock.advance(1000)
+        self.assertEquals(performerChosen, [])
+
+        # Enable processing
+        qpool.enable()
+
+        clock.advance(1000)
+        self.assertEquals(performerChosen, [True])
+
+        # Wait for job
+        while (yield inTransaction(dbpool.pool.connection, lambda txn: JobItem.all(txn))):
+            clock.advance(1)
+
+        # Work item complete
+        self.assertTrue(DummyWorkItem.results == {1: 12})
+
+
+
 class HalfConnection(object):
     def __init__(self, protocol):
         self.protocol = protocol
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <https://lists.macosforge.org/pipermail/calendarserver-changes/attachments/20150514/852c3749/attachment.html>