Consuming given strem object and returns processing stats.

        :param stream: streaming object to consume
        :type stream: iterable
        :param source: source of stream to consume
        :type source: string
        :param chunksize: chunk size for multiprocessing
        :type chunksize: integer
        :rtype: dict

def consume(self, stream, source=None, chunksize=1):
        """ Consuming given strem object and returns processing stats.

        :param stream: streaming object to consume
        :type stream: iterable
        :param source: source of stream to consume
        :type source: string
        :param chunksize: chunk size for multiprocessing
        :type chunksize: integer
        :rtype: dict
        """
        stats = {
            PROCESSING_TOTAL: 0,
            PROCESSING_SKIPPED: 0,
            PROCESSING_SUCCESS: 0,
            PROCESSING_ERROR: 0
        }
        if source:
            stats['source'] = source

        def skip_unless(r):
            if r:
                return r
            stats[PROCESSING_SKIPPED] += 1
            stats[PROCESSING_TOTAL] += 1

        rs = ifilter(skip_unless, stream)
        if self.processes:
            pool = multiprocessing.Pool(processes=self.processes)
            for f in self.procedures:
                rs = pool.imap_unordered(f, ifilter(skip_unless, rs),
                        chunksize=chunksize)
        else:
            for f in self.procedures:
                rs = imap(f, ifilter(skip_unless, rs))
        start = time.time()
        i = 0
        try:
            while 1:
                processed = next(rs)
                if processed is None:
                    stats[PROCESSING_SKIPPED] += 1
                elif processed is False:
                    stats[PROCESSING_ERROR] += 1
                else:
                    stats[PROCESSING_SUCCESS] += 1
                    self.collect(processed)
                i += 1
                stats[PROCESSING_TOTAL] += 1
                if i % self.reporting_interval == 0:
                    logging.info(" ===> Processed %dth item <=== ", i)
        except StopIteration:
            pass
        except KeyboardInterrupt:
            logging.info("Stopped by user interruption at %dth item.", i)
            raise
        except:
            e = sys.exc_info()[1]
            logging.error(e)
            raise
        finally:
            if self.processes:
                pool.close()
                pool.join()
            stats[PROCESSING_TIME] = time.time() - start
            logging.info(
                'STATS: total=%d, skipped=%d, success=%d, error=%d on %f[sec]'
                ' from "%s"',
                stats[PROCESSING_TOTAL], stats[PROCESSING_SKIPPED],
                stats[PROCESSING_SUCCESS], stats[PROCESSING_ERROR],
                stats[PROCESSING_TIME], stats.get('source', 'unknown'))
            return stats