def work(self):
        consumed = 0
        batch = []
        fingerprints = set()
        for m in self.consumer.get_messages(count=self.consumer_batch_size, timeout=1.0):
            try:
                msg = self._decoder.decode(m)
            except (KeyError, TypeError), e:
                logger.error("Decoding error: %s", e)
                continue
            else:
                type = msg[0]
                batch.append(msg)
                if type == 'add_seeds':
                    _, seeds = msg
                    fingerprints.update(map(lambda x: x.meta['fingerprint'], seeds))
                    continue

                if type == 'page_crawled':
                    _, response, links = msg
                    fingerprints.add(response.meta['fingerprint'])
                    fingerprints.update(map(lambda x: x.meta['fingerprint'], links))
                    continue

                if type == 'request_error':
                    _, request, error = msg
                    fingerprints.add(request.meta['fingerprint'])
                    continue

                if type == 'offset':
                    continue
                raise TypeError(type % 'Unknown message type %s')
            finally:
                consumed += 1

        self.states.fetch(fingerprints)
        fingerprints.clear()
        results = []
        for msg in batch:
            if len(results) > 1024:
                self.scoring_log_producer.send(None, *results)
                results = []

            type = msg[0]
            if type == 'add_seeds':
                _, seeds = msg
                for seed in seeds:
                    seed.meta['jid'] = self.job_id
                results.extend(self.on_add_seeds(seeds))
                continue

            if type == 'page_crawled':
                _, response, links = msg
                if response.meta['jid'] != self.job_id:
                    continue
                results.extend(self.on_page_crawled(response, links))
                continue

            if type == 'request_error':
                _, request, error = msg
                if request.meta['jid'] != self.job_id:
                    continue
                results.extend(self.on_request_error(request, error))
                continue
        if len(results):
            self.scoring_log_producer.send(None, *results)

        if self.cache_flush_counter == 30:
            logger.info("Flushing states")
            self.states.flush(force_clear=False)
            logger.info("Flushing states finished")
            self.cache_flush_counter = 0

        self.cache_flush_counter += 1

        if self.strategy.finished():
            logger.info("Successfully reached the crawling goal. Exiting.")
            exit(0)

        logger.info("Consumed %d items.", consumed)
        self.stats['last_consumed'] = consumed
        self.stats['last_consumption_run'] = asctime()
dataset/ETHPy150Open scrapinghub/frontera/frontera/worker/strategy.py/StrategyWorker.work/swapped operands of `%`