Added ThreadedStatistics plugin class. Migrated DBStats to use ThreadedStatistics.

git-svn-id: https://svn.mcs.anl.gov/repos/bcfg/trunk/bcfg2@5889 ce84e21b-d406-0410-9b95-82705330c041

1 files changed, 31 insertions, 0 deletions

diff --git a/src/lib/Server/Plugin.py b/src/lib/Server/Plugin.py
index f3b97b336..ff87c15df 100644
--- a/src/lib/Server/Plugin.py
+++ b/src/lib/Server/Plugin.py
@@ -7,6 +7,8 @@ import lxml.etree
 import os
 import posixpath
 import re
+import Queue
+import threading
 
 from lxml.etree import XML, XMLSyntaxError
 
@@ -141,6 +143,35 @@ class Statistics(object):
     def process_statistics(self, client, xdata):
         pass
 
+class ThreadedStatistics(Statistics,
+                         threading.Thread):
+    '''Threaded statistics handling capability'''
+    def __init__(self, core, datastore):
+        Statistics.__init__(self)
+        threading.Thread.__init__(self)
+        # Event from the core signaling an exit
+        self.terminate = core.terminate
+        self.work_queue = Queue.Queue()
+        self.start()
+
+    def run(self):
+        while not (self.terminate.isSet() and self.work_queue.empty()):
+            try:
+                (xdata, client) = self.work_queue.get(block=True, timeout=5)
+            except Queue.Empty:
+                continue
+            except Exception, e:
+                logger.error("ThreadedStatistics: %s" % e)
+                continue
+            self.handle_statistic(xdata, client)
+
+    def process_statistics(self, metadata, data):
+        self.work_queue.put((metadata, copy.deepcopy(data)))
+
+    def handle_statistics(self, metadata, data):
+        '''Handle stats here'''
+        pass
+
 class PullSource(object):
     def GetExtra(self, client):
         return []