ERSSubscriber.ERSSubscriber Class Reference

Public Member Functions
	__init__ (self, config)
str	default_id (self)
bool	add_callback (self, function, name, selection='.*')
	clear_callbacks (self)
bool	remove_callback (self, name)
	start (self)
	stop (self)
	message_loop (self)

Public Attributes
	bootstrap = config["bootstrap"]
str	group = config["group_id"]
	timeout = config["timeout"]
bool	running = False
	functions = dict()
	thread = threading.Thread(target=self.message_loop)

Detailed Description

Definition at line 15 of file ERSSubscriber.py.

Constructor & Destructor Documentation

__init__()

ERSSubscriber.ERSSubscriber.__init__	(		self,
			config )

Definition at line 16 of file ERSSubscriber.py.

    def __init__(self, config) :
        self.bootstrap = config["bootstrap"]
        if ( 'group_id' in config ) :
            self.group = config["group_id"]
        else:
            self.group = ""
        self.timeout = config["timeout"]
        self.running = False
        self.functions = dict()
        self.thread = threading.Thread(target=self.message_loop)
 
                        

Member Function Documentation

add_callback()

bool ERSSubscriber.ERSSubscriber.add_callback	(		self,
			function,
			name,
			selection = '.*' )

Definition at line 38 of file ERSSubscriber.py.

    def add_callback(self, function, name, selection  = '.*') -> bool:
        if ( name in self.functions ) : return False
       
        was_running = self.running
        if (was_running) : self.stop()
        
        prog = re.compile(selection)
        self.functions[name] = [prog, function]
 
        if (was_running) : self.start()
        return True
 

clear_callbacks()

ERSSubscriber.ERSSubscriber.clear_callbacks

(

self

)

Definition at line 50 of file ERSSubscriber.py.

    def clear_callbacks(self):
        if ( self.running ) :
            self.stop()
        self.functions.clear()
 

default_id()

str ERSSubscriber.ERSSubscriber.default_id

(

self

)

Definition at line 30 of file ERSSubscriber.py.

    def default_id(self) -> str:  
        node = socket.gethostname()
        user = getpass.getuser()
        process = os.getpid()
        thread = threading.get_ident()
        id = "{}-{}-{}-{}".format(node, user, process, thread)
        return id
           

message_loop()

ERSSubscriber.ERSSubscriber.message_loop

(

self

)

Definition at line 74 of file ERSSubscriber.py.

    def message_loop(self) :
        if (self.group == ""): group_id = self.default_id()
        else: group_id = self.group
 
        consumer = KafkaConsumer(bootstrap_servers=self.bootstrap,
                                 group_id=group_id, 
                                 client_id=self.default_id(),
                                 consumer_timeout_ms=self.timeout)
        
        topics = ["ers_stream"]
        consumer.subscribe(["monitoring." + s for s in topics])
 
        print("ID:", group_id, "running with functions:", *self.functions.keys())
 
        while ( self.running ) :
            try:
                message_it = iter(consumer)
                message = next(message_it)
                timestamp = message.timestamp
                key = message.key.decode('ascii')
                
                
                for function in self.functions.values() :
                    if function[0].match(key) :
                        issue = ersissue.IssueChain()
                        issue.ParseFromString( message.value )
                        function[1](issue)
                
            except msg.DecodeError :
                print("Could not parse message")
            except StopIteration :
                pass
            except Exception as e:
                print(e)
 
        print ("Stop")
 
 
        

remove_callback()

bool ERSSubscriber.ERSSubscriber.remove_callback	(		self,
			name )

Definition at line 55 of file ERSSubscriber.py.

    def remove_callback(self, name) -> bool:
        if ( name not in sef.functions.keys() ) : return False
 
        was_running = self.running
        if (was_running) : self.stop()
 
        self.functions.pop(name)
 
        if ( was_running and len(self.functions)>0 ) : self.start()
        return True

start()

ERSSubscriber.ERSSubscriber.start

(

self

)

Definition at line 65 of file ERSSubscriber.py.

    def start(self):
        print("Starting run")
        self.running = True
        self.thread.start()
 

stop()

ERSSubscriber.ERSSubscriber.stop

(

self

)

Definition at line 70 of file ERSSubscriber.py.

    def stop(self) :
        self.running = False
        self.thread.join()
 

Member Data Documentation

bootstrap

ERSSubscriber.ERSSubscriber.bootstrap = config["bootstrap"]

Definition at line 17 of file ERSSubscriber.py.

functions

ERSSubscriber.ERSSubscriber.functions = dict()

Definition at line 24 of file ERSSubscriber.py.

group

str ERSSubscriber.ERSSubscriber.group = config["group_id"]

Definition at line 19 of file ERSSubscriber.py.

running

bool ERSSubscriber.ERSSubscriber.running = False

Definition at line 23 of file ERSSubscriber.py.

thread

ERSSubscriber.ERSSubscriber.thread = threading.Thread(target=self.message_loop)

Definition at line 25 of file ERSSubscriber.py.

timeout

ERSSubscriber.ERSSubscriber.timeout = config["timeout"]

Definition at line 22 of file ERSSubscriber.py.

---

The documentation for this class was generated from the following file:

• /github/workspace/dunedaq/sourcecode/erskafka/python/erskafka/ERSSubscriber.py