[oddments/cappie.git] / cappie / queue.py

#!/usr/bin/python
###############################################################################
#                                                                             #
# Cappie                                                                      #
# Copyright (C) 2010 Michael Tremer                                           #
#                                                                             #
# This program is free software: you can redistribute it and/or modify        #
# it under the terms of the GNU General Public License as published by        #
# the Free Software Foundation, either version 3 of the License, or           #
# (at your option) any later version.                                         #
#                                                                             #
# This program is distributed in the hope that it will be useful,             #
# but WITHOUT ANY WARRANTY; without even the implied warranty of              #
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the               #
# GNU General Public License for more details.                                #
#                                                                             #
# You should have received a copy of the GNU General Public License           #
# along with this program.  If not, see <http://www.gnu.org/licenses/>.       #
#                                                                             #
###############################################################################

import time

from threading import Thread

import util

from database import Database
from errors import *
from events import *

class Queue(Thread):
	heartbeat = 1.0
	maxitems = 10000

	def __init__(self, log):
		Thread.__init__(self)

		self.log = log

		self.__running = True
		self.__queue = []

		self.db = Database(log)
		self.lastgc = None

	def __len__(self):
		return self.length

	def add(self, event):
		if self.length > self.maxitems:
			raise QueueFullError, "Cannot queue new event."

		self.__queue.append(event)

	@property
	def length(self):
		return len(self.__queue)

	def run(self):
		self.log.debug("Started event queue")

		util.setprocname("queue")

		self.db.open()

		while self.__running or self.__queue:
			if not self.__queue:
				#self.log.debug("Queue sleeping for %s seconds" % self.heartbeat)
				time.sleep(self.heartbeat)
				continue

			self._checkGc()

			event = self.__queue.pop(0)
			self.log.debug("Processing queue event: %s" % event)
			try:
				event.run()
			except EventException, e:
				self.log.error("Catched event exception: %s" % e)

		self.db.close()

	def shutdown(self):
		self.__running = False
		self.log.debug("Shutting down queue")
		self.log.debug("%d events in queue left" % len(self.__queue))

		# Wait until queue handled all events
		self.join()

	def _checkGc(self):
		if not self.lastgc or self.lastgc <= (time.time() - DB_GC_INTERVAL):
			self.add(EventGarbageCollector(self.db, self.log))
			self.lastgc = time.time()
Commit	Line	Data
53478050 MT	1	#!/usr/bin/python
	2	###############################################################################
	3	# #
	4	# Cappie #
	5	# Copyright (C) 2010 Michael Tremer #
	6	# #
	7	# This program is free software: you can redistribute it and/or modify #
	8	# it under the terms of the GNU General Public License as published by #
	9	# the Free Software Foundation, either version 3 of the License, or #
	10	# (at your option) any later version. #
	11	# #
	12	# This program is distributed in the hope that it will be useful, #
	13	# but WITHOUT ANY WARRANTY; without even the implied warranty of #
	14	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the #
	15	# GNU General Public License for more details. #
	16	# #
	17	# You should have received a copy of the GNU General Public License #
	18	# along with this program. If not, see <http://www.gnu.org/licenses/>. #
	19	# #
	20	###############################################################################
	21
	22	import time
	23
	24	from threading import Thread
	25
64e40ed0 MT	26	import util
64e40ed0 MT	27
af1f4801	28	from database import Database
53478050	29	from errors import *
4eaf4298	30	from events import *
53478050 MT	31
	32	class Queue(Thread):
	33	heartbeat = 1.0
4eaf4298	34	maxitems = 10000
53478050 MT	35
	36	def __init__(self, log):
	37	Thread.__init__(self)
	38
	39	self.log = log
	40
	41	self.__running = True
	42	self.__queue = []
	43
af1f4801	44	self.db = Database(log)
4eaf4298	45	self.lastgc = None
af1f4801	46
53478050 MT	47	def __len__(self):
	48	return self.length
	49
	50	def add(self, event):
	51	if self.length > self.maxitems:
	52	raise QueueFullError, "Cannot queue new event."
	53
	54	self.__queue.append(event)
	55
	56	@property
	57	def length(self):
	58	return len(self.__queue)
	59
	60	def run(self):
	61	self.log.debug("Started event queue")
	62
64e40ed0 MT	63	util.setprocname("queue")
64e40ed0 MT	64
af1f4801 MT	65	self.db.open()
af1f4801 MT	66
53478050 MT	67	while self.__running or self.__queue:
	68	if not self.__queue:
	69	#self.log.debug("Queue sleeping for %s seconds" % self.heartbeat)
	70	time.sleep(self.heartbeat)
	71	continue
	72
4eaf4298 MT	73	self._checkGc()
4eaf4298 MT	74
53478050 MT	75	event = self.__queue.pop(0)
	76	self.log.debug("Processing queue event: %s" % event)
	77	try:
	78	event.run()
	79	except EventException, e:
	80	self.log.error("Catched event exception: %s" % e)
	81
af1f4801 MT	82	self.db.close()
af1f4801 MT	83
53478050 MT	84	def shutdown(self):
	85	self.__running = False
	86	self.log.debug("Shutting down queue")
	87	self.log.debug("%d events in queue left" % len(self.__queue))
	88
	89	# Wait until queue handled all events
	90	self.join()
4eaf4298 MT	91
	92	def _checkGc(self):
	93	if not self.lastgc or self.lastgc <= (time.time() - DB_GC_INTERVAL):
	94	self.add(EventGarbageCollector(self.db, self.log))
	95	self.lastgc = time.time()