1#!/router/bin/python
2
3import os
4import outer_packages
5import zmq
6import threading
7import logging
8import CCustomLogger
9import zipmsg
10from json import JSONDecoder
11from common.trex_status_e import TRexStatus
12
13# setup the logger
14CCustomLogger.setup_custom_logger('TRexServer')
15logger = logging.getLogger('TRexServer')
16
17
18class ZmqMonitorSession(threading.Thread):
19    def __init__(self, trexObj , zmq_port):
20        super(ZmqMonitorSession, self).__init__()
21        self.stoprequest    = threading.Event()
22        self.first_dump     = True
23        self.zmq_port       = zmq_port
24        self.zmq_publisher  = "tcp://localhost:{port}".format(port=self.zmq_port)
25        self.trexObj        = trexObj
26        self.expect_trex    = self.trexObj.expect_trex     # used to signal if TRex is expected to run and if data should be considered
27        self.decoder        = JSONDecoder()
28        self.zipped         = zipmsg.ZippedMsg()
29        logger.info("ZMQ monitor initialization finished")
30
31    def run(self):
32        try:
33            self.context    = zmq.Context()
34            self.socket     = self.context.socket(zmq.SUB)
35            logger.info("ZMQ monitor started listening @ {pub}".format(pub=self.zmq_publisher))
36            self.socket.connect(self.zmq_publisher)
37            self.socket.setsockopt(zmq.SUBSCRIBE, b'')
38
39            while not self.stoprequest.is_set():
40                try:
41                    zmq_dump = self.socket.recv()   # This call is BLOCKING until data received!
42                    if self.expect_trex.is_set():
43                        self.parse_and_update_zmq_dump(zmq_dump)
44                        logger.debug("ZMQ dump received on socket, and saved to trexObject.")
45                except Exception as e:
46                    if self.stoprequest.is_set():
47                        # allow this exception since it comes from ZMQ monitor termination
48                        pass
49                    else:
50                        logger.error("ZMQ monitor thrown an exception. Received exception: {ex}".format(ex=e))
51                        self.trexObj.zmq_error = e
52        except Exception as e:
53            logger.error('ZMQ monitor error: %s' % e)
54            self.trexObj.zmq_error = e
55
56    def join(self, timeout=5):
57        self.stoprequest.set()
58        logger.debug("Handling termination of ZMQ monitor thread")
59        self.socket.close()
60        self.context.term()
61        logger.info("ZMQ monitor resources has been freed.")
62        super(ZmqMonitorSession, self).join(timeout)
63
64    def parse_and_update_zmq_dump(self, zmq_dump):
65        unzipped = self.zipped.decompress(zmq_dump)
66        if unzipped:
67            zmq_dump = unzipped
68        dict_obj = self.decoder.decode(zmq_dump.decode(errors = 'replace'))
69
70        if type(dict_obj) is not dict:
71            raise Exception('Expected ZMQ dump of type dict, got: %s' % type(dict_obj))
72
73        # add to trex_obj zmq latest dump, based on its 'name' header
74        if dict_obj != {}:
75            self.trexObj.update_zmq_dump_key(dict_obj['name'], dict_obj)
76            if self.first_dump:
77                # change TRexStatus from starting to Running once the first ZMQ dump is obtained and parsed successfully
78                self.first_dump = False
79                self.trexObj.set_status(TRexStatus.Running)
80                self.trexObj.set_verbose_status("TRex is Running")
81                logger.info("First ZMQ dump received and successfully parsed. TRex running state changed to 'Running'.")
82
83
84if __name__ == "__main__":
85    pass
86
87