openrvdas.logger.writers.timeout_writer

  1#!/usr/bin/env python3
  2
  3import threading
  4import time
  5
  6from logger.writers.writer import Writer  # noqa: E402
  7
  8
  9class TimeoutWriter(Writer):
 10    def __init__(self, writer, timeout, message=None, resume_message=None,
 11                 empty_is_okay=False, none_is_okay=False, **kwargs):
 12        """Instantiated with a client Writer instance (such as a
 13        LogfileWriter), an interval, a timeout and optional
 14        message. Expects its write() method to be called at least every
 15        'timeout' seconds. If it isn't, call the client's write method
 16        with 'message' to indicate that it has timed out. Once it receives
 17        a call to its write() method after timing out, call the client's
 18        write method with 'resume_message' to indicate that it is no
 19        longer timed out.
 20        ```
 21        writer         A client writer instance
 22
 23        timeout        Timeout interval in seconds
 24
 25        message        Message to be returned if client reader fails to return
 26                       a record within the timeout interval
 27
 28        resume_message Message to be returned when client returns a record after
 29                       having timed out
 30
 31        empty_is_okay If True, receiving an empty record is sufficient to reset
 32                      the timer.
 33        none_is_okay  If True, receiving a 'None' record is sufficient to reset
 34                        the timer.
 35        ```
 36        Sample config that echos stdin and issues timeouts if no input for 5 secs:
 37        ```
 38          readers:
 39          - class: TextFileReader
 40          transforms:
 41          - class: TimestampTransform
 42          writers:
 43          - class: TextFileWriter
 44          - class: TimeoutWriter
 45            kwargs:
 46              writer:
 47                class: TextFileWriter
 48              timeout: 5
 49              message: No message received for 5 seconds
 50              resume_message: Okay, got another message
 51        ```
 52        """
 53        # Initialize type checking
 54        super().__init__(**kwargs)  # processes 'quiet' and type hints
 55
 56        self.writer = writer
 57        self.timeout = timeout
 58        self.message = message or ('Timeout: no %s record received in %d seconds'
 59                                   % (writer, timeout))
 60        self.resume_message = resume_message or ('Timeout: %s record received'
 61                                                 % writer)
 62        self.empty_is_okay = empty_is_okay
 63        self.none_is_okay = none_is_okay
 64
 65        # When we got our last record (or were instantiated)
 66        self.last_record = time.time()
 67
 68        # Keep track of whether we're currently timed out or not
 69        self.timed_out = False
 70
 71        # Protect self.last_record and self.timed_out
 72        self.timeout_lock = threading.Lock()
 73
 74        # To let us cleanly exit _timeout_thread
 75        self.quit_signaled = False
 76
 77        # Start the timeout loop in a separate thread
 78        self.timeout_thread = threading.Thread(target=self._timeout_thread,
 79                                               name='timeout_thread', daemon=True)
 80        self.timeout_thread.start()
 81
 82    ############################
 83    def __del__(self):
 84        self.quit()
 85
 86    ############################
 87    def _timeout_thread(self):
 88        """Call client write() if we have/haven't had our own write() called
 89        within the alloted time.
 90        """
 91        while not self.quit_signaled:
 92            now = time.time()
 93            with self.timeout_lock:
 94                time_to_sleep = self.timeout - (now - self.last_record)
 95
 96                # If we're overdue for a record...
 97                if time_to_sleep < 0:
 98                    # If we weren't already timed out, we are now. Send message.
 99                    if not self.timed_out:
100                        self.timed_out = True
101                        self.writer.write(self.message)
102
103                    # Check again in timeout seconds
104                    time_to_sleep = self.timeout
105
106            # Whether or not we're timed out, snooze until we expect our
107            # next timeout.
108            time.sleep(time_to_sleep)
109
110    ############################
111    def quit(self):
112        self.quit_signaled = True
113
114    ############################
115    def write(self, record):
116        """Register that we've had a write() call; reset timeout timer."""
117        if record is None and not self.none_is_okay:
118            return
119        if not record and not self.empty_is_okay:
120            return
121
122        # If here, we got a bona fide record. Reset our timer.
123        with self.timeout_lock:
124            if self.timed_out:
125                self.timed_out = False
126                self.writer.write(self.resume_message)
127            self.last_record = time.time()