<commit_before>require 'datagram_worker/version'
require 'logger'
require 'bunny'
require 'json'
module DatagramWorker

  def self.subscribe(routing_key, logger = ::Logger.new(STDOUT),&block)
    if ENV['RABBITMQ_URL']
      logger.info("RabbitMQ Connecting to #{ENV['RABBITMQ_URL']}")
      $conn = Bunny.new(ENV['RABBITMQ_URL'])
    else
      logger.info("RabbitMQ connecting to localhost")
      $conn = Bunny.new
    end

    $conn.start

    $ch = $conn.create_channel
    $x  = $ch.topic('datagrams_topic_exchange', auto_delete: false)
    $datagram_responses =  $ch.queue("datagram_responses", :durable => true)

    $ch.queue('', durable: true).bind($x, routing_key: routing_key).subscribe(block: true) do |di, md, pl|
      yield(di,md,JSON.parse(pl))
    end


  end

end<commit_msg>smoothens response sending<commit_after>require 'datagram_worker/version'
require 'logger'
require 'bunny'
require 'json'
module DatagramWorker

  def self.subscribe(routing_key, logger = ::Logger.new(STDOUT),&block)
    @@responses ||= {}
    if ENV['RABBITMQ_URL']
      logger.info("RabbitMQ Connecting to #{ENV['RABBITMQ_URL']}")
      $conn = Bunny.new(ENV['RABBITMQ_URL'])
    else
      logger.info("RabbitMQ connecting to localhost")
      $conn = Bunny.new
    end

    $conn.start

    $ch = $conn.create_channel
    $x  = $ch.topic('datagrams_topic_exchange', auto_delete: false)
    $datagram_responses =  $ch.queue("watch_responses", :durable => true)

    $ch.queue('', durable: true).bind($x, routing_key: routing_key).subscribe(block: true) do |di, md, pl|
      logger.info(pl)
      j = JSON.parse(pl)
      @@responses[j["key"]] = Time.now
      yield(di,md,j)
    end
  end

  def self.respond_with(responses)
    ap @@responses
    responses.each do |id, response|
      r = {
        status_code: 200,
        data: {
          status_code: 200,
          data: response,
          elapsed: Time.now - @@responses[id]
        },
        id: id
      }
      @@responses.delete(id)
      $datagram_responses.publish(r.to_json)

    end

  end


end
{'fraction_non_alphanumeric': 0.14008725157537566, 'fraction_numerical': 0.00290838584585555, 'mean_word_length': 3.2208588957055215, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '7449572', 'n_tokens_mistral': 762, 'n_tokens_neox': 713, 'n_words': 142}