1

"""

     2

    Non-blocking fifo reads.

     3

"""

     4

     5

import os

     6

import errno

     7

import fcntl

     8

     9

import logging

    10

    11

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

    12

    13

class Pipe (object) :

    14

    """

    15

        A pipe from a fd.

    16

    17

        Supports reading lines in a non-blocking fashion.

    18

    """

    19

    20

    @classmethod

    21

    def file (cls, file) :

    22

        """

    23

            Create Pipe from file, e.g. sys.stdin.

    24

    25

            Puts fd into nonblocking mode, which means that the given file will stop working!

    26

        """

    27

    28

        fd = file.fileno()

    29

    30

        log.debug("%s: %s", file, fd)

    31

    32

        fl = fcntl.fcntl(fd, fcntl.F_GETFL)

    33

        fl |= os.O_NONBLOCK

    34

        fcntl.fcntl(fd, fcntl.F_SETFL, fl)

    35

    36

        return cls(fd)

    37

    38

    def __init__ (self, fd) :

    39

        self._fd = fd

    40

        self._buf = ''

    41

    42

        log.debug("pipe: %d", fd)

    43

    44

    def fileno (self) :

    45

        """

    46

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

    47

        """

    48

    49

        if self._fd is None :

    50

            raise ValueError("I/O operation on closed pipe: %s" % (self, ))

    51

        else :

    52

            return self._fd

    55

    def read (self, n=512) :

    56

        """

    57

            Read up to n bytes.

    58

    59

            Returns None if we would block.

    60

            Raises EOFError on EOF.

    61

        """

    62

    63

        try :

    64

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

    65

    66

        except OSError as ex :

    67

            # block?

    68

            if ex.errno == errno.EAGAIN :

    69

                # empty

    71

    72

            else :

    73

                raise

    74

    80

            return buf

    81

        else :

    82

            raise EOFError()

    83

    84

    def readline (self) :

    85

        """

    86

            Read and return next waiting line from input.

    87

    88

            Line is returned without trailing '\n'.

    89

    90

            Returns None if there is no line available.

    91

            Raises EOFError if the fifo write end was closed.

    92

        """

    93

    94

        while '\n' not in self._buf :

    95

            # read chunk

    96

            read = self.read()

    97

    98

            if read is None :

    99

                return None

   100

   101

            self._buf += read

   102

   103

        # split out one line

   104

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

   105

   106

        log.debug("%s", line)

   107

   108

        return line

   109

   110

    def readlines (self) :

   111

        """

   112

            Read any available input, yielding lines.

   113

   114

            Re-opens the FIFO on EOF.

   115

   116

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

   117

        """

   118

   119

        while True :

   120

            # pull line

   121

            line = self.readline()

   122

   123

            if line :

   124

                yield line

   125

            else :

   126

                return # block

   127

   128

    __iter__ = readlines

   129

   130

    def __str__ (self) :

   131

        return "pipe:{self._fd}".format(self=self)

   132

   133

class Fifo (Pipe) :

   134

    """

   135

        A named pipe(7) on the filesystem.

   136

   137

        Supports reading lines in a non-blocking fashion, and re-opening on EOF.

   138

    """

   139

   140

    def __init__ (self, path) :

   141

        self.path = path

   142

        Pipe.__init__(self, self._open())

   143

   144

    def _open (self) :

   145

        """

   146

            Open the internal fd (nonblocking).

   147

        """

   148

   149

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

   150

   151

        log.debug("%s: open: %s", self, fd)

   152

   153

        return fd

   154

   155

    def _close (self, fd) :

   156

        """

   157

            Close.

   158

        """

   159

   160

        os.close(fd)

   161

   162

    def _reopen (self) :

   163

        """

   164

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

   165

        """

   166

   167

        self._close(self._fd)

   168

        self._fd = None

   169

        self._fd = self._open()

   170

   171

    def readlines (self) :

   172

        """

   173

            Read any available input, yielding lines.

   174

   175

            Re-opens the FIFO on EOF.

   176

   177

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

   178

        """

   179

   180

        while True :

   181

            try :

   182

                # pull line

   183

                line = self.readline()

   184

   185

            except EOFError :

   186

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

   187

   188

                # reopen and go back to waiting

   189

                self._reopen()

   190

   191

                return

   192

   193

            if line is None :

   194

                log.debug("%s: EOF: wait", self)

   195

                return # wait

   196

            else :

   197

                yield line

   198

   199

    __iter__ = readlines

   200

   201

    def close (self) :

   202

        """

   203

            Close the fifo.

   204

        """

   205

   206

        self._close(self.fileno())

   207

        self._fd = None

   208

   209

    def __del__ (self) :

   210

        """

   211

            Cleanup

   212

        """

   213

   214

        if self._fd is not None :

   215

            self._close(self._fd)

   216

            self._fd = None

   217