Coverage for melissa/utility/timer.py: 89%

1#!/usr/bin/python3 

2 

3# Copyright (c) 2021-2022, Institut National de Recherche en Informatique et en Automatique (Inria) 

4# All rights reserved. 

5# 

6# Redistribution and use in source and binary forms, with or without 

7# modification, are permitted provided that the following conditions are met: 

8# 

9# * Redistributions of source code must retain the above copyright notice, this 

10# list of conditions and the following disclaimer. 

11# 

12# * Redistributions in binary form must reproduce the above copyright notice, 

13# this list of conditions and the following disclaimer in the documentation 

14# and/or other materials provided with the distribution. 

15# 

16# * Neither the name of the copyright holder nor the names of its 

17# contributors may be used to endorse or promote products derived from 

18# this software without specific prior written permission. 

19# 

20# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 

21# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 

22# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE 

23# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE 

24# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL 

25# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR 

26# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER 

27# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, 

28# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE 

29# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. 

30 

31"""This module defines a `Timer` class for sending a signal signals 

32to a socket with a timed loops.""" 

33 

34import select 

35import socket 

36 

37from melissa.utility import time 

38from melissa.utility.networking import Socket 

39 

40 

41class Timer: 

42 """ 

43 Timer class that runs a timed loop, sending a signal through a socket after 

44 a specified interval if no activity is detected on the socket. 

45 

46 Attributes: 

47 _fd (Socket): The socket file descriptor to monitor. 

48 _interval (time.Time): The interval duration to wait for activity. 

49 """ 

50 

51 def __init__(self, fd: Socket, interval: time.Time) -> None: 

52 """ 

53 Initializes the Timer with a socket file descriptor and an interval. 

54 

55 Args: 

56 fd (Socket): The socket to monitor. 

57 interval (time.Time): The time interval to wait for activity. 

58 

59 Raises: 

60 AssertionError: If the interval is not greater than zero. 

61 """ 

62 assert interval.total_seconds() > 0 

63 

64 self._fd = fd 

65 self._interval = interval 

66 

67 def run(self) -> None: 

68 """ 

69 Starts the timer and waits for activity on the socket. If no activity 

70 is detected within the interval, a null byte is sent through the socket 

71 to avoid generating a SIGPIPE. If the socket is closed, it handles the 

72 BrokenPipeError gracefully. 

73 

74 If activity is detected on the socket, the method returns. 

75 

76 Raises: 

77 BrokenPipeError: If the socket is closed before sending. 

78 """ 

79 while True: 

80 rs, _, _ = select.select( 

81 [self._fd], [], [], self._interval.total_seconds() 

82 ) 

83 

84 if rs == []: 

85 # No activity detected, send a signal to prevent SIGPIPE 

86 try: 

87 self._fd.send(b'\0', socket.MSG_NOSIGNAL) 

88 except BrokenPipeError: 

89 pass 

90 else: 

91 assert rs[0] == self._fd 

92 return