self.daemon = True

        filelike = False

        self.bytes = _bytes()

        if type(source) in (type(''), _bytes, _bytearray):  # string-like

            self.bytes = _bytes(source)

        else:  # can be either file pointer or file-like

            if type(source) in (int, long):  # file pointer it is

                ## converting file descriptor (int) stdin into file-like

                source = os.fdopen(source, 'rb', 16384)

            # let's see if source is file-like by now

            filelike = hasattr(source, 'read')

        if not filelike and not self.bytes:

            raise TypeError("StreamFeeder's source object must be a readable "

                            "file-like, a file descriptor, or a string-like.")

        self.source = source

        self.readiface, self.writeiface = os.pipe()

    def run(self):

        t = self.writeiface

        if self.bytes:

            os.write(t, self.bytes)

        else:

            s = self.source

            b = s.read(4096)

            while b:

                os.write(t, b)

                b = s.read(4096)

        os.close(t)

    @property

    def output(self):

        return self.readiface

class InputStreamChunker(Thread):

    def __init__(self, source, target, buffer_size, chunk_size):

        super(InputStreamChunker, self).__init__()

        self.daemon = True  # die die die.

        self.source = source

        self.target = target

        self.chunk_count_max = int(buffer_size / chunk_size) + 1

        self.chunk_size = chunk_size

        self.data_added = Event()

        self.data_added.clear()

        self.keep_reading = Event()

        self.keep_reading.set()

        self.EOF = Event()

        self.EOF.clear()

        self.go = Event()

        self.go.set()

    def stop(self):

        self.go.clear()

        self.EOF.set()

        try:

            # this is not proper, but is done to force the reader thread let

            # go of the input because, if successful, .close() will send EOF

            # down the pipe.

            self.source.close()

        except:

            pass

    def run(self):

        s = self.source

        t = self.target

        cs = self.chunk_size

        ccm = self.chunk_count_max

        kr = self.keep_reading

        da = self.data_added

        go = self.go

        try:

            b = s.read(cs)

        except ValueError:

            b = ''

        while b and go.is_set():

            if len(t) > ccm:

                kr.clear()

                kr.wait(2)

                # # this only works on 2.7.x and up

                # if not kr.wait(10):

                #     raise Exception("Timed out while waiting for input to be read.")

                # instead we'll use this

                if len(t) > ccm + 3:

                    raise IOError(

                        "Timed out while waiting for input from subprocess.")

            t.append(b)

            da.set()

            b = s.read(cs)

        self.EOF.set()

        da.set()  # for cases when done but there was no input.

class BufferedGenerator(object):

    """

    Class behaves as a non-blocking, buffered pipe reader.

    Reads chunks of data (through a thread)

    from a blocking pipe, and attaches these to an array (Deque) of chunks.

    Reading is halted in the thread when max chunks is internally buffered.

    The .next() may operate in blocking or non-blocking fashion by yielding

    '' if no data is ready

    to be sent or by not returning until there is some data to send

    When we get EOF from underlying source pipe we raise the marker to raise

    StopIteration after the last chunk of data is yielded.

    """

    def __init__(self, source, buffer_size=65536, chunk_size=4096,

                 starting_values=[], bottomless=False):

        if bottomless:

            maxlen = int(buffer_size / chunk_size)

        else:

            maxlen = None

        self.data = deque(starting_values, maxlen)

        self.worker = InputStreamChunker(source, self.data, buffer_size,

                                         chunk_size)

        if starting_values:

            self.worker.data_added.set()

        self.worker.start()

    ####################

    # Generator's methods

    ####################

    def __iter__(self):

        return self

    def next(self):

        while not len(self.data) and not self.worker.EOF.is_set():

            self.worker.data_added.clear()

            self.worker.data_added.wait(0.2)

        if len(self.data):

            self.worker.keep_reading.set()

            return _bytes(self.data.popleft())

        elif self.worker.EOF.is_set():

            raise StopIteration

    def throw(self, type, value=None, traceback=None):

        if not self.worker.EOF.is_set():

            raise type(value)

    def start(self):

        self.worker.start()

    def stop(self):

        self.worker.stop()

    def close(self):

        try:

            self.worker.stop()

            self.throw(GeneratorExit)

        except (GeneratorExit, StopIteration):

            pass

    def __del__(self):

        self.close()

    ####################

    # Threaded reader's infrastructure.

    ####################

    @property

    def input(self):

        return self.worker.w

    @property

    def data_added_event(self):

        return self.worker.data_added

    @property

    def data_added(self):

        return self.worker.data_added.is_set()

    @property

    def reading_paused(self):

        return not self.worker.keep_reading.is_set()

    @property

    def done_reading_event(self):

        """

        Done_reding does not mean that the iterator's buffer is empty.

        Iterator might have done reading from underlying source, but the read

        chunks might still be available for serving through .next() method.

        :returns: An Event class instance.