<commit_before>import capturer from 'chrome-har-capturer'
import log from './debug'
import analyze from './analysis/analyze'

export default function createWorker (opts) {
  const { cache, rankings, chromeConfig, interval, ttl, verbose } = opts
  const worker = {
    async processQueue () {
      const url = await cache.rpop('queue')
      cache.publish(`queue-pop:${url}`, null)
      if (url === null) {
        return setTimeout(worker.processQueue, interval)
      } else {
        try {
          await worker.analyzeUrl(url)
        } catch (err) {
          log.error(err)
        }
        cache.publish(`queue-next`, null)
        return setImmediate(worker.processQueue)
      }
    },
    async analyzeUrl (url) {
      const handleError = (err) => {
        log.debug(err)
        cache.publish(`analysis-error:${url}`, null)
      }
      // Start loading URL in chrome
      const har = await new Promise((resolve) => {
        capturer.setVerbose(verbose)
        capturer.load([url], chromeConfig)
          .on('error', handleError)
          .on('pageError', handleError)
          .on('connect', function () {
            cache.publish(`har-start:${url}`, null)
          })
          .on('end', function (data) {
            const json = JSON.stringify(data)
            cache.publish(`har-done:${url}`, json)
            resolve({ url: url, data: data, json: json })
          })
      })
      cache.publish(`analysis-start:${url}`, null)<commit_msg>Use redis BRPOP instead of RPOP - use blocking behavior instead of pulling<commit_after>import capturer from 'chrome-har-capturer'
import log from './debug'
import analyze from './analysis/analyze'

export default function createWorker (opts) {
  const { cache, rankings, chromeConfig, interval, ttl, verbose } = opts
  const worker = {
    async processQueue () {
      // it blocks the connection indefinitely when there are no elements
      const result = await cache.brpop('queue', 0)
      if (result) {
        const url = result[1]
        cache.publish(`queue-pop:${url}`, null)
        try {
          await worker.analyzeUrl(url)
        } catch (err) {
          log.error(err)
        }
        cache.publish(`queue-next`, null)
        return worker.processQueue()
      } else {
        return setTimeout(worker.processQueue, interval)
      }
    },
    async analyzeUrl (url) {
      const handleError = (err) => {
        log.debug(err)
        cache.publish(`analysis-error:${url}`, null)
      }
      // Start loading URL in chrome
      const har = await new Promise((resolve) => {
        capturer.setVerbose(verbose)
        capturer.load([url], chromeConfig)
          .on('error', handleError)
          .on('pageError', handleError)
          .on('connect', function () {
            cache.publish(`har-start:${url}`, null)
          })
          .on('end', function (data) {
            const json = JSON.stringify(data)
            cache.publish(`har-done:${url}`, json)
            resolve({ url: url, data: data, json: json })
          })
      })
      cache.publish(`analysis-start:${url}`, null)
{'fraction_non_alphanumeric': 0.13530927835051546, 'fraction_numerical': 0.0006443298969072165, 'mean_word_length': 2.220954356846473, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4072225', 'n_tokens_mistral': 889, 'n_tokens_neox': 875, 'n_words': 221}