pandajedi/jedimsgprocessor/base_msg_processor.py

0001 from pandacommon.pandamsgbkr.msg_processor import SimpleMsgProcPluginBase
0002
0003 from pandajedi.jedicore.JediTaskBuffer import JediTaskBuffer
0004
0005
0006 # Base simple message processing plugin
0007 class BaseMsgProcPlugin(SimpleMsgProcPluginBase):
0008     def initialize(self, in_collective=False):
0009         """
0010         initialize plugin instance, run once before loop in thread
0011         """
0012         if in_collective:
0013             # run in collective msg_proc plugin
0014             pass
0015         else:
0016             # run as individual msg_proc plugin
0017             self.set_tbIF()
0018
0019     def set_tbIF(self):
0020         """
0021         set up JEDI TaskBuffer interface
0022         """
0023         # set nDBConnection = n_threads to avoid DBProxy blocking amongs threads
0024         n_db_conns = getattr(self, "n_threads", 1)
0025         self.tbIF = JediTaskBuffer(None, nDBConnection=n_db_conns)
0026
0027     def process(self, msg_obj):
0028         """
0029         process the message
0030         Get msg_obj from the incoming MQ (if any; otherwise msg_obj is None)
0031         Returned value will be sent to the outgoing MQ (if any)
0032         """
0033         pass