<commit_before>require File.expand_path(File.join(File.dirname(__FILE__), 'publisher'))
require 'singleton'

module Dboard
  class Collector
    include Singleton

    attr_reader :sources

    def self.register_source(key, instance)
      Collector.instance.register_source(key, instance)
    end

    def self.register_after_update_callback(callback)
      Collector.instance.register_after_update_callback(callback)
    end

    def self.start
      instance.start
    end

    def initialize
      @sources = {}
      @after_update_callback = lambda {}
    end

    def start
      @sources.each do |source, instance|
        Thread.new do
          loop do
            update_in_thread(source, instance)
          end
        end
      end
      loop { sleep 1 }
    end

    def update_in_thread(source, instance)
      time = Time.now
      puts "#{source} updating..."
      update_source(source, instance)
      elapsed_time = Time.now - time
      time_until_next_update = instance.update_interval - elapsed_time
      time_until_next_update = 0 if time_until_next_update < 0
      puts "#{source} done in #{elapsed_time} seconds, will update again in #{time_until_next_update} seconds (interval: #{instance.update_interval})."
      sleep time_until_next_update
    rescue Exception => ex
      puts "Something failed outside the update_source method. #{ex.message}"
      puts ex.backtrace
    end

    def update_source(source, instance)
      begin
        data = instance.fetch
        publish_data(source, data)
      ensure
        @after_update_callback.call
      end
    rescue Exception => ex
      puts "Failed to update #{source}: #{ex.message}"
      puts ex.backtrace
    end

    def register_source(key, instance)
      @sources.merge!({ key => instance })
    end

    def register_after_update_callback(callback)
      @after_update_callback = callback
    end

    def publish_data(source, data)
      Publisher.publish(source, data)
    end
  end
end<commit_msg>Spreading out the load when booting the dashboard daemon.<commit_after>require File.expand_path(File.join(File.dirname(__FILE__), 'publisher'))
require 'singleton'

module Dboard
  class Collector
    include Singleton

    attr_reader :sources

    def self.register_source(key, instance)
      Collector.instance.register_source(key, instance)
    end

    def self.register_after_update_callback(callback)
      Collector.instance.register_after_update_callback(callback)
    end

    def self.start
      instance.start
    end

    def initialize
      @sources = {}
      @after_update_callback = lambda {}
    end

    def start
      @sources.each do |source, instance|
        wait_a_little_bit_to_not_start_all_fetches_at_once

        Thread.new do
          loop do
            update_in_thread(source, instance)
          end
        end
      end
      loop { sleep 1 }
    end

    def update_in_thread(source, instance)
      time = Time.now
      puts "#{source} updating..."
      update_source(source, instance)
      elapsed_time = Time.now - time
      time_until_next_update = instance.update_interval - elapsed_time
      time_until_next_update = 0 if time_until_next_update < 0
      puts "#{source} done in #{elapsed_time} seconds, will update again in #{time_until_next_update} seconds (interval: #{instance.update_interval})."
      sleep time_until_next_update
    rescue Exception => ex
      puts "Something failed outside the update_source method. #{ex.message}"
      puts ex.backtrace
    end

    def update_source(source, instance)
      begin
        data = instance.fetch
        publish_data(source, data)
      ensure
        @after_update_callback.call
      end
    rescue Exception => ex
      puts "Failed to update #{source}: #{ex.message}"
      puts ex.backtrace
    end

    def register_source(key, instance)
      @sources.merge!({ key => instance })
    end

    def register_after_update_callback(callback)
      @after_update_callback = callback
    end

    def publish_data(source, data)
      Publisher.publish(source, data)
    end

    def wait_a_little_bit_to_not_start_all_fetches_at_once
      sleep 10 * rand
    end
  end
end
{'fraction_non_alphanumeric': 0.07464114832535886, 'fraction_numerical': 0.0019138755980861245, 'mean_word_length': 2.753141831238779, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 9, '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': '1825450', 'n_tokens_mistral': 1222, 'n_tokens_neox': 1195, 'n_words': 306}