<commit_before># @(#) $Id: Consumer.py,v 1.16 2005/10/24 11:40:56 dfugate Exp $
#
# Copyright (C) 2001
# Associated Universities, Inc. Washington DC, USA.
#
# Produced for the ALMA project
#
# This library is free software; you can redistribute it and/or modify it under
# the terms of the GNU Library General Public License as published by the Free
# Software Foundation; either version 2 of the License, or (at your option) any
# later version.
#
# This library is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU Library General Public License for more
# details.
#
# You should have received a copy of the GNU Library General Public License
# along with this library; if not, write to the Free Software Foundation, Inc.,
# 675 Massachusetts Ave, Cambridge, MA 02139, USA.  Correspondence concerning
# ALMA should be addressed as follows:
#
# Internet email: <EMAIL>
#------------------------------------------------------------------------------
'''
This module includes classes to be used as Consumers for the CORBA Notification
service.
'''

__revision__ = "$Id: Consumer.py,v 1.16 2005/10/24 11:40:56 dfugate Exp $"

#--REGULAR IMPORTS-------------------------------------------------------------
from traceback import print_exc
#--CORBA STUBS-----------------------------------------------------------------
import CosNotifyChannelAdmin
import CosNotifyComm__POA
import CosNotification
import CosNotifyFilter
import acsnc
from ACSErrTypeCommonImpl         import CORBAProblemExImpl
from ACSErr                       import NameValue
#--ACS Imports-----------------------------------------------------------------
from Acspy.Common.Log         import getLogger
from Acspy.Nc.CommonNC        import CommonNC
from Acspy.Nc.CDBProperties   import get_integration_logs
#--GLOBALS---------------------------------------------------------------------

#------------------------------------------------------------------------------
class Consumer (CosNotifyComm__POA.StructuredPushConsumer, CommonNC):
    #--------------------------------------------------------------------------
    '''
    Class Consumer is the implementation of a CORBA Structured Push Consumer.
    It is provided to hide details of the CORBA Notification Service from
    the developer.  It can be instantiated as-is or it can be subclassed and
    the developer should override the processEvent method.
    '''
    #--------------------------------------------------------------------------
    def __init__ (self, name, component=None):
        '''
        Constructor.

        Params:
        - name is the channel name in string format
        - component is the component this supplier has been instantiated from
        (if applicable). This parameter is likely to become mandatory in future
        version of ACS

        Returns: Nothing

        Raises: ACSErrTypeCommonImpl.CORBAProblemExImpl on critical failures
        '''
        self.logger = getLogger(str(name) + "-Consumer")

        #Call the super's constructor
        CommonNC.__init__(self, name, component)
        
        #CORBA ref to the channels consumer admin
        self.consumerAdmin = None  
        #CORBA ref to this consumers structured proxy push supplier
        self.spps = None
        #Dictionary containing handler functions for different object types
        self.handlers = {}  
       
        #Handle all of the CORBA stuff now
        CommonNC.initCORBA(self)
        self.initCORBA()
    #--------------------------------------------------------------------------
    def initCORBA(self):
        '''
        Handles all the CORBA involved in creating a Consumer.

        Parameters: None

        Returns: Nothing

        Raises: ACSErrTypeCommonImpl.CORBAProblemExImpl on critical failures
        '''
            
        #Get admin object. 
        try:
            # get admin object for the consumer and the proxy supplier from
            # admin
            (self.consumerAdmin, adminid) = self.evtChan.new_for_consumers(CosNotifyChannelAdmin.AND_OP)
            (self.spps, pid) = self.consumerAdmin.obtain_notification_push_supplier(CosNotifyChannelAdmin.STRUCTURED_EVENT)
            #unfortunately this HAS to be narrowed.
            self.spps = self.spps._narrow(CosNotifyChannelAdmin.StructuredProxyPushSupplier)

            # to make NRI happy
            pid = 0
            adminid = 0
            
        except Exception, e:
            print_exc()
            raise CORBAProblemExImpl(nvSeq=[NameValue("channelname",
                                                      self.channelName),
                                            NameValue("reason",
                                                      "Bad proxy or admin"),
                                            NameValue("exception", str(e))])
        return
    #--------------------------------------------------------------------------
    def push_structured_event (self, event):
        '''
        CORBA method that is invoked by suppliers. If a correct handler function
        has been provided via the addSubscription method, this is used to do
        something useful with the event. Furthermore, if the user-defined handler
        function raises an exception, this is caught and the processEvent
        method is also invoked. Otherwise, its up to the developer to subclass
        Consumer and override the processEvent method.

        Parameters: event is a CosNotification.StructuredEvent
        
        Returns: Nothing

        Raises: Nothing
        '''
        #For HLA/ITS - to be removed later!
        if get_integration_logs(self.channelName)==1:
            try:
                comp_name = self.component.getName()
            except:
                comp_name = "Unknown"
                
            self.logger.logNotice("Receiver:" + comp_name +
                                  ", Event Type:" + event.header.fixed_header.event_type.type_name)
        
        #check to see if the developer has provided a handler first
        if self.handlers.has_key(event.header.fixed_header.event_type.type_name):
            try:
                #get the function
                temp_func = self.handlers[event.header.fixed_header.event_type.type_name]

                #convert the CORBA any into a "normal" Python object
                real_obj = event.filterable_data[0].value
                real_obj = real_obj.value()

                #invoke the user-defined function on it
                temp_func(real_obj)

                #ignore everything else
                return
            
            except Exception, e:
                self.logger.logCritical('Unable to use handler function...' +
                                        str(e))
                print_exc()

        #either a handler is not used or it failed...pass the se to
        #processEvent(...) and hope the developer has overriden that method.
        try:
            self.processEvent(type_name=event.header.fixed_header.event_type.type_name,
                              event_name=event.header.fixed_header.event_name,
                              corba_any=event.filterable_data[0].value,
                              se=event)
        except Exception, e:
            self.logger.logCritical('processEvent(...)...' + str(e))
            print_exc()

    #--------------------------------------------------------------------------
    def suspend(self):
        '''<commit_msg>Events not processed within 2 seconds (or some amount of time specified within in the CDB) now produce nasty warning log.<commit_after># @(#) $Id: Consumer.py,v 1.17 2006/03/03 22:29:49 dfugate Exp $
#
# Copyright (C) 2001
# Associated Universities, Inc. Washington DC, USA.
#
# Produced for the ALMA project
#
# This library is free software; you can redistribute it and/or modify it under
# the terms of the GNU Library General Public License as published by the Free
# Software Foundation; either version 2 of the License, or (at your option) any
# later version.
#
# This library is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU Library General Public License for more
# details.
#
# You should have received a copy of the GNU Library General Public License
# along with this library; if not, write to the Free Software Foundation, Inc.,
# 675 Massachusetts Ave, Cambridge, MA 02139, USA.  Correspondence concerning
# ALMA should be addressed as follows:
#
# Internet email: <EMAIL>
#------------------------------------------------------------------------------
'''
This module includes classes to be used as Consumers for the CORBA Notification
service.
'''

__revision__ = "$Id: Consumer.py,v 1.17 2006/03/03 22:29:49 dfugate Exp $"

#--REGULAR IMPORTS-------------------------------------------------------------
from traceback import print_exc
#--CORBA STUBS-----------------------------------------------------------------
import CosNotifyChannelAdmin
import CosNotifyComm__POA
import CosNotification
import CosNotifyFilter
import acsnc
from ACSErrTypeCommonImpl         import CORBAProblemExImpl
from ACSErr                       import NameValue
#--ACS Imports-----------------------------------------------------------------
from Acspy.Common.Log         import getLogger
from Acspy.Nc.CommonNC        import CommonNC
from Acspy.Nc.CDBProperties   import get_integration_logs
from Acspy.Nc.CDBProperties   import getEventHandlerTimeoutDict
from Acspy.Util.Profiler      import Profiler
#--GLOBALS---------------------------------------------------------------------
#the default maximum amount of time an event handler is given to process event
#before an exception is logged. this is used when an enduser does *not* define
#the appropriate XML elements within the ACS CDB. see the inline doc on EventChannel.xsd
#for more info
DEFAULT_MAX_PROCESS_TIME = 2.0

#------------------------------------------------------------------------------
class Consumer (CosNotifyComm__POA.StructuredPushConsumer, CommonNC):
    #--------------------------------------------------------------------------
    '''
    Class Consumer is the implementation of a CORBA Structured Push Consumer.
    It is provided to hide details of the CORBA Notification Service from
    the developer.  It can be instantiated as-is or it can be subclassed and
    the developer should override the processEvent method.
    '''
    #--------------------------------------------------------------------------
    def __init__ (self, name, component=None):
        '''
        Constructor.

        Params:
        - name is the channel name in string format
        - component is the component this supplier has been instantiated from
        (if applicable). This parameter is likely to become mandatory in future
        version of ACS

        Returns: Nothing

        Raises: ACSErrTypeCommonImpl.CORBAProblemExImpl on critical failures
        '''
        self.logger = getLogger(str(name) + "-Consumer")

        #Call the super's constructor
        CommonNC.__init__(self, name, component)
        
        #CORBA ref to the channels consumer admin
        self.consumerAdmin = None  
        #CORBA ref to this consumers structured proxy push supplier
        self.spps = None
        #Dictionary containing handler functions for different object types
        self.handlers = {}
        #diction mapping events to the total amount of time they have
        #to finish
        self.handlerTimeoutDict = getEventHandlerTimeoutDict(name)
        #profiler to time how long it takes events to be processed
        self.profiler = Profiler()
       
        #Handle all of the CORBA stuff now
        CommonNC.initCORBA(self)
        self.initCORBA()
    #--------------------------------------------------------------------------
    def initCORBA(self):
        '''
        Handles all the CORBA involved in creating a Consumer.

        Parameters: None

        Returns: Nothing

        Raises: ACSErrTypeCommonImpl.CORBAProblemExImpl on critical failures
        '''
            
        #Get admin object. 
        try:
            # get admin object for the consumer and the proxy supplier from
            # admin
            (self.consumerAdmin, adminid) = self.evtChan.new_for_consumers(CosNotifyChannelAdmin.AND_OP)
            (self.spps, pid) = self.consumerAdmin.obtain_notification_push_supplier(CosNotifyChannelAdmin.STRUCTURED_EVENT)
            #unfortunately this HAS to be narrowed.
            self.spps = self.spps._narrow(CosNotifyChannelAdmin.StructuredProxyPushSupplier)

            # to make NRI happy
            pid = 0
            adminid = 0
            
        except Exception, e:
            print_exc()
            raise CORBAProblemExImpl(nvSeq=[NameValue("channelname",
                                                      self.channelName),
                                            NameValue("reason",
                                                      "Bad proxy or admin"),
                                            NameValue("exception", str(e))])
        return
    #--------------------------------------------------------------------------
    def push_structured_event (self, event):
        '''
        CORBA method that is invoked by suppliers. If a correct handler function
        has been provided via the addSubscription method, this is used to do
        something useful with the event. Furthermore, if the user-defined handler
        function raises an exception, this is caught and the processEvent
        method is also invoked. Otherwise, its up to the developer to subclass
        Consumer and override the processEvent method.

        Parameters: event is a CosNotification.StructuredEvent
        
        Returns: Nothing

        Raises: Nothing
        '''
        #For HLA/ITS - to be removed later!
        if get_integration_logs(self.channelName)==1:
            try:
                comp_name = self.component.getName()
            except:
                comp_name = "Unknown"
                
            self.logger.logNotice("Receiver:" + comp_name +
                                  ", Event Type:" + event.header.fixed_header.event_type.type_name)

        #For HLA - maximum amount of time the consumer has to process the event
        if self.handlerTimeoutDict.has_key(event.header.fixed_header.event_type.type_name) == 0:
            self.handlerTimeoutDict[event.header.fixed_header.event_type.type_name] = DEFAULT_MAX_PROCESS_TIME

        max_process_time = self.handlerTimeoutDict[event.header.fixed_header.event_type.type_name]
        
        #check to see if the developer has provided a handler first
        if self.handlers.has_key(event.header.fixed_header.event_type.type_name):
            try:
                #get the function
                temp_func = self.handlers[event.header.fixed_header.event_type.type_name]

                #convert the CORBA any into a "normal" Python object
                real_obj = event.filterable_data[0].value
                real_obj = real_obj.value()

                #invoke the user-defined function on it. must time it
                self.profiler.start()
                temp_func(real_obj)
                time_to_run = self.profiler.stop() / 1000.0
                self.profiler.reset()

                if time_to_run > max_process_time:
                    self.logger.logCritical("Took too long to handle an '" +
                                            event.header.fixed_header.event_type.type_name +
                                            "' event: " + str(time_to_run) + " seconds.")
                    self.logger.logInfo("Maximum time to process an event is: " + str(max_process_time) + " seconds.")

                #ignore everything else
                return
            
            except Exception, e:
                self.logger.logCritical('Unable to use handler function...' +
                                        str(e))
                print_exc()

        #either a handler is not used or it failed...pass the se to
        #processEvent(...) and hope the developer has overriden that method.
        try:
            self.profiler.start()
            
            self.processEvent(type_name=event.header.fixed_header.event_type.type_name,
                              event_name=event.header.fixed_header.event_name,
                              corba_any=event.filterable_data[0].value,
                              se=event)

            time_to_run = self.profiler.stop() / 1000.0
            self.profiler.reset()

            if time_to_run > max_process_time:
                self.logger.logCritical("Took too long to handle an '" +
                                        event.header.fixed_header.event_type.type_name +
                                        "' event: " + str(time_to_run) + " seconds.")
                self.logger.logInfo("Maximum time to process an event is: " + str(max_process_time) + " seconds.")
            
        except Exception, e:
            self.logger.logCritical('processEvent(...)...' + str(e))
            print_exc()

    #--------------------------------------------------------------------------
    def suspend(self):
        '''
{'fraction_non_alphanumeric': 0.15027744408214633, 'fraction_numerical': 0.006750185916137521, 'mean_word_length': 1.5727740986019132, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 27, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4223550', 'n_tokens_mistral': 4210, 'n_tokens_neox': 3926, 'n_words': 1388}