1

"""

     2

    Non-blocking fifo reads.

     3

"""

     4

     5

import os

     6

import errno

     7

import select

     8

     9

import logging

    10

    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

    42

    def _open (self) :

    43

        """

    44

            Open the internal fd (nonblocking).

    45

        """

    46

    47

        assert self._fd is None

    48

    49

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

    50

    51

    def _close (self) :

    52

        """

    53

            Close.

    54

        """

    55

    56

        assert self._fd is not None

    57

    58

        os.close(self._fd)

    59

    60

        self._fd = None

    61

    62

    def _reopen (self) :

    63

        """

    64

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

    65

        """

    66

    67

        self._close()

    68

        self._open()

    69

    70

    def poll (self, timeout=None) :

    71

        """

    72

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

    73

    74

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

    75

    76

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

    77

        """

    78

    79

        if timeout is False :

    80

            timeout = 0.0

    81

    82

        # select

    83

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

    84

    85

        if read :

    86

            return True

    87

    88

        else :

    89

            # timeout

    90

            return False

    91

    92

    def read (self, n=512) :

    93

        """

    94

            Read up to n bytes.

    95

    96

            Returns None if we would block.

    97

            Raises EOFError on EOF.

    98

        """

    99

   100

        try :

   101

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

   102

   103

        except OSError as ex :

   104

            # block?

   105

            if ex.errno == errno.EAGAIN :

   106

                # empty

   107

                return None

   108

   109

            else :

   110

                raise

   111

   112

        # eof?

   113

        if not buf :

   114

            raise EOFError()

   115

   116

        # ok

   117

        return buf

   118

   119

    def readline (self) :

   120

        """

   121

            Read and return next waiting line from input.

   122

   123

            Line is returned without trailing '\n'.

   124

   125

            Returns None if there is no line available.

   126

            Raises EOFError if the fifo write end was closed.

   127

        """

   128

   129

        while '\n' not in self._buf :

   130

            # read chunk

   131

            read = self.read()

   132

   133

            if read is None :

   134

                return None

   135

   136

            self._buf += read

   137

   138

        # split out one line

   139

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

   140

   141

        return line

   142

   143

    def readlines (self) :

   144

        """

   145

            Read any available input, yielding lines.

   146

   147

            Re-opens the FIFO on EOF.

   148

   149

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

   150

        """

   151

   152

        while True :

   153

            try :

   154

                # pull line

   155

                line = self.readline()

   156

   157

            except EOFError :

   158

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

   159

   160

                # reopen and go back to waiting

   161

                self._reopen()

   162

   163

                return

   164

   165

            if line is None :

   166

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

   167

   168

                # wait

   169

                return

   170

   171

            log.debug("%s", line)

   172

            yield line

   173

   174

    __iter__ = readlines

   175

   176

    def close (self) :

   177

        """

   178

            Close the fifo.

   179

        """

   180

   181

        if self._fd is None :

   182

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

   183

   184

        self._close()

   185

   186

    def __del__ (self) :

   187

        """

   188

            Cleanup

   189

        """

   190

   191

        # will not close stdin

   192

        if self._fd :

   193

            self._close()