openrvdas.logger.listener.listener

  1#!/usr/bin/env python3
  2
  3import logging
  4import logging.handlers
  5import time
  6import traceback
  7
  8
  9from logger.readers.composed_reader import ComposedReader  # noqa: E402
 10from logger.writers.composed_writer import ComposedWriter  # noqa: E402
 11
 12
 13################################################################################
 14class Listener:
 15    """Listener is a simple, yet relatively self-contained class that
 16    takes a list of one or more Readers, a list of zero or more
 17    Transforms, and a list of zero or more Writers. It calls the Readers
 18    (in parallel) to acquire records, passes those records through the
 19    Transforms (in series), and sends the resulting records to the Writers
 20    (in parallel).
 21
 22    """
 23    ############################
 24
 25    def __init__(self, readers=[], transforms=[], writers=[], stderr_writers=[],
 26                 host_id='', interval=0, name=None):
 27        """listener = Listener(readers, transforms=[], writers=[],
 28                            interval=0)
 29
 30        readers        A single Reader or a list of Readers.
 31
 32        transforms     A single Transform or a list of zero or more Transforms
 33
 34        writers        A single Writer or a list of zero or more Writers
 35
 36        stderr_writers A single Writer or a list of zero or more Writers to which
 37                       the logger's stderr should be written.
 38
 39        interval       How long to sleep before reading sequential records
 40
 41        name           Optional human-readable short name to be used in displays
 42
 43        Sample use:
 44
 45        listener = Listener(readers=[NetworkReader(':6221'),
 46                                     NetworkReader(':6223')],
 47                            transforms=[TimestampTransform()],
 48                            writers=[TextFileWriter('/logs/network_recs'),
 49                                     TextFileWriter(None)],
 50                            interval=0.2)
 51        listener.run()
 52
 53        Calling listener.quit() from another thread will cause the run() loop
 54        to exit.
 55        """
 56        logging.info('Instantiating %s logger', name or 'unnamed')
 57
 58        ###########
 59        # Create readers, writers, etc.
 60        self.reader = ComposedReader(readers=readers)
 61        self.writer = ComposedWriter(transforms=transforms, writers=writers)
 62        self.interval = interval
 63        self.name = name or 'Unnamed listener'
 64        self.last_read = 0
 65
 66        self.quit_signalled = False
 67
 68    ############################
 69    def quit(self):
 70        """
 71        Signal 'quit' to all the readers.
 72        """
 73        self.quit_signalled = True
 74        logging.info('Shutting down %s', self.name)
 75
 76    ############################
 77    def run(self):
 78        """
 79        Read/transform/write until either quit() is called in a separate
 80        thread, or ComposedReader returns None, indicating that all its
 81        component readers have returned EOF.
 82        """
 83        logging.info('Running %s', self.name)
 84
 85        if not self.reader and not self.writer:
 86            logging.info('No readers or writers defined - exiting.')
 87            return
 88
 89        record = ''
 90        try:
 91            while not self.quit_signalled and record is not None:
 92                record = self.reader.read()
 93                self.last_read = time.time()
 94                logging.debug('ComposedReader read: "%s"', record)
 95                if record:
 96                    self.writer.write(record)
 97
 98                if self.interval:
 99                    time_to_sleep = self.interval - (time.time() - self.last_read)
100                    time.sleep(max(time_to_sleep, 0))
101
102        # Exit in an orderly fashion if someone hits Ctl-C
103        except KeyboardInterrupt:
104            logging.info('Listener %s received KeyboardInterrupt - exiting.',
105                         self.name or '')
106        except Exception as e:
107            logging.info('Listener %s received exception: %s',
108                         self.name, traceback.format_exc())
109            raise e