C:/work/RevealedSystems/EyeSpy-RS/eyespy/src/objects/storage/storageManager.py

Go to the documentation of this file.

# Copyright (C) 2008-2009  Mark Hanegraaff
#
# 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 wx
from sqlite3Impl import sqlite3Impl
from objects.logic.network.packetDecoder import packetDecoder
from threading import  RLock, Thread, Event
from objects.logic.common.exception import exception
from objects.logic.common.logger import logger
from objects.logic.common.globalStrings import globalStrings
import time

from objects.logic.common.threads.appThreadBase import appThreadBase
from objects.logic.common.observerInterface import observerInterface


## The storageWriterThread class is a long lived worked thread used by the storageManager class, whose job is
# to write a copy of all captured data to the SQLite database
class storageWriterThread(appThreadBase):

    def run ( self ):
        try:
            self.parent.storageImpl.open()
            self.anythingElseProcessed = False
            logger.log(__name__ + ": Starting Storage Write Thread")
            while(self.killSwitch == False):
                self.threadEvent.clear()
                                
                try:
                    
                    try:
                        (header, data) = self.parent.packetList.pop(0)
                    except Exception:
                        time.sleep(1)
                        
                    else:
                        decodedPacketDict = self.parent.decoder.decodePacket(header, data)
                        self.parent.storageImpl.store(decodedPacketDict)
                        
                        self.anythingElseProcessed = True
                            
                        
                    if self.parent.timeToCommit == True and self.anythingElseProcessed == True:
                        self.parent.timeToCommit = False
                        self.parent.anythingElseProcessed = False
                         
                        self.parent.storageImpl.commit()
                        
                except Exception, ex:
                    logger.log(__name__ + ": There was an unknown error writing packet to database")
                    logger.log(ex)            
                finally:
                    pass            
 
                                        
        finally:
            self.parent.storageImpl.commit()
            self.parent.storageImpl.close()
            self.threadEvent.set()
            self.threadEvent.clear()
            logger.log(__name__ + ": Storage Write Thread is Done")
  



## The storageWriterThread class is a long lived worked thread used by the storageManager class, whose job is
# to do the following things:
# 1) set the commit flag periodically and
# 2) check to see that the storage file has not exceeded its size.
class storageMonitorThread(appThreadBase):
    
    def __init__(self, _parent, storageWriterThread):
        Thread.__init__(self)
        appThreadBase.__init__(self, _parent)
        
        self.storageWriterThread = storageWriterThread
        self.sleepSoFar = 0
        
    def run ( self ):
        logger.log(__name__ + ": Starting Storage Monitor Timer")
        self.firstTimeSizeCheck = True
        
        try:            
            
            while(self.killSwitch == False):
                
                self.threadEvent.clear()
                try:                    
                    time.sleep(3)
                    if self.parent.maxSize != -1:
                        self.sleepSoFar += 3                
                    self.parent.timeToCommit = True                
                                                            
                except Exception, ex:
                    logger.log(__name__ + ": There was an error setting storage commit flag. Data will now be commited on this cycle")
                    logger.log(ex)   
                    
                # could be expensive, so check every 30 seconds
                if self.firstTimeSizeCheck == True or self.sleepSoFar >= 30:
                    self.sleepSoFar = 0
                    self.firstTimeSizeCheck = False
                    
                        
                    fileSze = self.parent.storageImpl.getStorageSizeBytes()
                    if self.parent.maxSize != -1 and fileSze > (self.parent.maxSize * 1000000):
                        # kill the storage writer and monitor
                        self.storageWriterThread.killSignal()
                        self.killSwitch = True
                        
                        # alert the user that the storage file has exceeded its size
                        dlg = wx.MessageDialog(None, "The Trace Storage File has exceeded its maximum allowed size and no more data will be saved.\nTo increase capacity, set a new limit in the application options and restart the monitor", 'Error', wx.ICON_ERROR)
                        result = dlg.ShowModal()
                        dlg.Destroy()
                    else:
                       strMsg = "Storage File has not exceeded its max size yet -- Current size is [%d] Bytes" % fileSze
                       logger.log(__name__ + ": " + strMsg) 
                                            
        finally:
            self.threadEvent.set()
            self.threadEvent.clear()  
            logger.log(__name__ + ": Storage Monitor is Done")
            
            
            
            
## The storageManager class is responsible for somehow storing a copy of all the captured TCP/IP data
# into the application's SQLite database.
# This class also deals with other details, such as limiting the database file size to a user defined value, 
# and perform periodic commits.
class storageManager(observerInterface):
    
    def __init__(self, saveStorage, maxSize):        
        self.storageImpl = sqlite3Impl(globalStrings.traceStoragePath)
        
        self.decoder = packetDecoder()
        self.decoder.setDefaultReader()
        
        self.saveStorage = saveStorage
        self.maxSize     = maxSize
        self.lck = RLock()
        
        
        self.writer = storageWriterThread(self)
        self.monitor = storageMonitorThread(self, self.writer)
        
        self.packetList = []
        
        self.timeToCommit = False
        
        self.cnt = 0


    ## implementation of observer method. This class is an observer of the
    # networkMonitor Class.
    # @param header Raw DataLink header
    # @param data  Raw DataLink frame
    def update(self, header, data):        
        
        #self.cnt += 1
        #print "%d" % (self.cnt)
        
        try:
            self.processPacket(header, data)
        except exception, ex:
            logger.log(__name__ + ": Could add packet onto storage queue")
            logger.log(ex)
                   
                   
                   
    ## Initialized the internal storage file.
    def initializeStorage(self):
        self.storageImpl.createStorage()
        self.storageImpl.clear()
    
    
    ## Deletes the internal storage file.
    def deleteStorage(self):
        self.storageImpl.deleteStorage()
    
    
    
    ## Starts the storage threads.
    def startupStorage(self):
        if self.storageImpl.successful == True and self.saveStorage == True:
            self.__startThreads()
    
    
    
    ## Stops the storage threads
    def shutdownStorage(self):
        if self.storageImpl.successful == True and self.saveStorage == True:
            self.__stopThreads()            
            
    
    
    ## Processes a new packed. Simply adds the (header, data) tuple to a
    # list which is dequeued by the storageWriterThread class.
    # @param header Raw DataLink header
    # @param data  Raw DataLink frame
    def processPacket(self, header, data):
        
        if self.writer.getKillSignal() or self.monitor.getKillSignal():
            return
        
        try:
            self.packetList.append((header, data))
        finally:
            pass
    
    
    
    ## Helper function used to start worker threads.
    def __startThreads(self):
        if self.__threadIsRunning() == True:
            return
                
        self.writer = storageWriterThread(self)                
        self.monitor = storageMonitorThread(self, self.writer)
        
        self.writer.reset()
        self.monitor.reset()
    
        self.writer.start()
        self.monitor.start()
        
    
    
    ## Helper function used to stop worker threads.    
    def __stopThreads(self):
        
        if self.__threadIsRunning() == True:
                        
            self.writer.killSignal()       
            self.monitor.killSignal()
            
            self.writer.wait(10)       
            self.monitor.wait(10)
            
            if self.maxSize > 0:
                pass    
        
        
    ## Helper function used to check if threads are running.
    def __threadIsRunning(self):
        if self.writer != None and self.writer.isAlive() == True:
            return True
        else:
            return False
    
    
    

---