1

"""

     2

    Non-blocking fifo reads.

     3

"""

     4

     5

import os

     6

import errno

     7

import select

     8

     9

import logging

    10

    11

log = logging.getLogger('pvl.collectd.fifo')

    12

    13

class Fifo (object) :

    14

    """

    15

        A named pipe(7) on the filesystem.

    16

    17

        Supports reading lines in a non-blocking fashion.

    18

    """

    19

    20

    def __init__ (self, path) :

    21

        self.path = path

    22

        self._fd = None

    23

        self._buf = ''

    24

    25

        log.debug("open: %s", path)

    26

        self._open()

    27

    28

    @property

    29

    def fd (self) :

    30

        """

    31

            Fetch the internal fd, failing if we are not open..

    32

        """

    33

    34

        if self._fd is None :

    35

            raise ValueError("I/O operation on closed fifo: {0}".format(self.path))

    36

    37

        return self._fd

    38

    39

    def _open (self) :

    40

        """

    42

        """

    43

    44

        assert self._fd is None

    45

    46

        self._fd = os.open(self.path, os.O_RDONLY | os.O_NONBLOCK)

    47

    48

    def _close (self) :

    49

        """

    50

            Close.

    51

        """

    52

    53

        assert self._fd is not None

    54

    55

        os.close(self._fd)

    56

    57

        self._fd = None

    58

    59

    def _reopen (self) :

    60

        """

    61

            Re-open the FIFO in case the writing end was closed, and read gave EOF.

    62

        """

    63

    64

        self._close()

    65

        self._open()

    66

    67

    def poll (self, timeout=None) :

    68

        """

    69

            Poll for input, with given timeout in seconds (float).

    70

    71

            A timeout of None indicates to block forever, False indicates to never block.

    72

    73

            Returns True if we have input waiting, False on timeout with no input. None on indeterminate.

    74

        """

    75

    76

        if timeout is False :

    77

            timeout = 0.0

    78

    79

        # select

    80

        read, write, ex = select.select([self.fd], [], [], timeout)

    81

    82

        if read :

    83

            return True

    84

    85

        else :

    86

            # timeout

    87

            return False

    88

    89

    def read (self, n=512) :

    90

        """

    91

            Read up to n bytes.

    92

    93

            Returns None if we would block.

    94

            Raises EOFError on EOF.

    95

        """

    96

    97

        try :

    98

            buf = os.read(self.fd, n)

    99

   100

        except OSError as ex :

   101

            # block?

   102

            if ex.errno == errno.EAGAIN :

   103

                # empty

   104

                return None

   105

   106

            else :

   107

                raise

   108

   109

        # eof?

   110

        if not buf :

   111

            raise EOFError()

   112

   113

        # ok

   114

        return buf

   115

   116

    def readline (self) :

   117

        """

   118

            Read and return next waiting line from input.

   119

   120

            Line is returned without trailing '\n'.

   121

   122

            Returns None if there is no line available.

   123

            Raises EOFError if the fifo write end was closed.

   124

        """

   125

   126

        while '\n' not in self._buf :

   127

            # read chunk

   128

            read = self.read()

   129

   130

            if read is None :

   131

                return None

   132

   133

            self._buf += read

   134

   135

        # split out one line

   136

        line, self._buf = self._buf.split('\n', 1)

   137

   138

        return line

   139

   140

    def readlines (self) :

   141

        """

   142

            Read any available input, yielding lines.

   143

   144

            Re-opens the FIFO on EOF.

   145

   146

            Returns None if there was no more input available, or the fifo was re-opened after EOF.

   147

        """

   148

   149

        while True :

   150

            try :

   151

                # pull line

   152

                line = self.readline()

   153

   154

            except EOFError :

   155

                log.debug("EOF/reopen: %s", self.path)

   156

   157

                # reopen and go back to waiting

   158

                self._reopen()

   159

   160

                return

   161

   162

            if line is None :

   163

                log.debug("fifo empty: %s", self.path)

   164

   165

                # wait

   166

                return

   167

   168

            log.debug("%s", line)

   169

            yield line

   170

   171

    __iter__ = readlines

   172

   173

    def close (self) :

   174

        """

   175

            Close the fifo.

   176

        """

   177

   178

        if self._fd is None :

   179

            raise ValueError("Fifo already closed: {0}".format(self.path))

   180

   181

        self._close()

   182

   183

    def __del__ (self) :

   184

        """

   185

            Cleanup

   186

        """

   187

   188

        # will not close stdin

   189

        if self._fd :

   190

            self._close()