1

# read a stream from a fifo

     2

     3

from twisted.internet import reactor, interfaces

     4

from twisted.python import log

     5

from zope.interface import implements

     6

     7

import os, fcntl, errno

     8

     9

class EOF (Exception) : pass

    10

    11

BUF_SIZE = 2048

    12

    13

class Fifo (object) :

    14

    implements(interfaces.IReadDescriptor)

    15

    16

    def __init__ (self, path) :

    17

        self.path = path

    20

        self._open()

    21

    22

    def _open (self) :

    23

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

    24

    25

        reactor.addReader(self)

    26

    27

        log.msg("opened fifo %s as %d" % (self.path, self.fd))

    28

    29

    def _close (self) :

    30

        if self.fd :

    31

            reactor.removeReader(self)

    32

            os.close(self.fd)

    33

    34

            log.msg("closed fifo %d at %s" % (self.fd, self.path))

    35

    36

            self.fd = None

    37

    close = _close

    38

    39

    def reopen (self) :

    40

        """

    41

            Close and re-open the fifo. This is useful for handling EOF

    42

        """

    43

        self._close()

    44

        self._open()

    45

    46

    def _read (self, length) :

    47

    48

        log.msg("(read %d bytes from %d:%s)" % (length, self.fd, self.path))

    49

        try :

    50

            data = os.read(self.fd, length)

    51

    52

        except OSError, e :

    53

            if e.errno == errno.EAGAIN :

    54

                log.msg("\tEAGAIN")

    55

                return None

    56

            else :

    57

                log.msg("\tERROR: %s" % e)

    58

                raise

    59

    60

        if not data :

    61

            log.msg("\tEOF")

    62

            raise EOF()

    63

    64

        log.msg("\tDATA: %d: %r" % (len(data), data))

    65

        return data

    66

    67

    def fileno (self) :

    68

        return self.fd

    69

    70

    def doRead (self) :

    71

        while True :

    72

            log.msg("fifo doRead loop")

    73

    74

            try :

    75

                data = self._read(BUF_SIZE)

    76

            except EOF :

    77

                self.handleEOF()

    78

                return

    79

    80

            if data :

    81

                self.dataReceived(data)

    82

            else :

    83

                break

    84

    85

    def dataReceived (self, data) :

    86

        pass

    87

    88

    def handleEOF (self) :

    89

        pass

    90

    91

    def connectionLost (self, reason) :

    92

        self.close()

    93

    94

    def logPrefix (self) :

    95

        return "FIFO:%d:%s" % (self.fd, self.path)

    96

    97

    def __del__ (self) :

    98

        """

    99

            !!! this is important

   100

        """

   101

        self.close()

   102