s4z / plumd

            err = "attempt to remove non-existant render: {0}"

            self.log.error(err.format(exc))

    def run(self):

        """Start reading from the reader Plugin after calling its onstart.

        Call stop() to get this thread to exit - before returning it will

        call the readers onstop().

        """

        if not self.queue or not self.render:

            err = "output queue/render not set for reader: {0}"

            raise PluginRuntimeError(err.format(self.plugin.name))

        # starting up, call onstart

        self.plugin.onstart()

        # distribute metrics - don't everyone flood all at once

        time.sleep(random.randrange(int(self.start_delay)))

        # save a few cycles

        tloop = self.plugin.interval

        evt = self.stop_evt

        poll = self.plugin.poll

        # loop on calling the plugin objects poll and sending to writers

        while not evt.is_set():

            # loop every self.interval seconds

            with Interval(tloop, evt):

                # poll for metric result set

                rset = poll()

                # readers can return None values eg. a reader that polls

                # very often and periodically returns results

                if rset is None:

                    continue

                # write the value to each of our output queues directly

                for render in self.renders:

                    render.process(rset.results)

        # all done, call onstop

        self.plugin.onstop()

class WriterThread(PluginThread):

        for i in range(0, 10):

            self.queue.put_nowait(None)

    def run(self):

        """Start writing metrics to self.writer from self.queue.

        First calls self.writer.onstart().

        Call stop() to get this thread to exit - before returning it will

        call the writers onstop().

        """

        if not self.queue or not self.render:

            err = "input queue/render not set for writer: {0}"

            raise PluginRuntimeError(err.format(self.plugin.name))

        # starting up, call onstart

        self.plugin.onstart()

        # save a few cycles

        stop_evt = self.stop_evt

        write = self.plugin.write

        # loop on popping from queue and calling self.push()

        while not stop_evt.is_set():

            entry = self.queue.get()

            if entry:

                write(entry)

        # flush any partially rendered metrics on shutdown

        partial = self.render.flush()

        if partial:

            self.plugin.flush(partial)

        # all done, call onstop

        self.plugin.onstop()