<commit_before>require 'spec_helper'
require 'yaml'
require 'lagomorph/session'
require 'lagomorph/supervisor'
require 'lagomorph/subscriber'
require 'lagomorph/rpc_call'
require 'lagomorph/exceptions'

require 'pong_worker'

describe 'a Lagomorph RPC process' do

  let(:rabbitmq_config) {
    YAML.load_file(
      File.join(File.expand_path(File.dirname(__FILE__)),'rabbitmq.yml')
    )
  }

  context 'when supervising a PongWorker on the ping queue' do
    let(:connection_params) {
      {
        host:     rabbitmq_config.fetch('host'),
        username: rabbitmq_config.fetch('username'),
        password: <PASSWORD>.fetch('password')
      }
    }

    let(:session) { Lagomorph::Session.connect(connection_params) }

    let(:supervisor) { Lagomorph::Supervisor.new(session) }

    let(:queue) { 'ping' }
<commit_msg>Add spec for multiple subscriber concurrency<commit_after>require 'spec_helper'
require 'yaml'
require 'lagomorph/session'
require 'lagomorph/supervisor'
require 'lagomorph/subscriber'
require 'lagomorph/rpc_call'
require 'lagomorph/exceptions'

require 'pong_worker'

describe 'a Lagomorph RPC process' do

  let(:rabbitmq_config) {
    YAML.load_file(
      File.join(File.expand_path(File.dirname(__FILE__)),'rabbitmq.yml')
    )
  }

  context 'when supervising a PongWorker on the ping queue' do
    let(:connection_params) {
      {
        host:     rabbitmq_config.fetch('host'),
        username: rabbitmq_config.fetch('username'),
        password: <PASSWORD>config.fetch('password')
      }
    }

    let(:session) { Lagomorph::Session.connect(connection_params) }

    let(:supervisor) { Lagomorph::Supervisor.new(session) }

    let(:queue) { 'ping' }

    context 'with tasks of varying completion time' do
      let!(:rpc_call) { Lagomorph::RpcCall.new(session) }
      let!(:results) { [] }

      before do
        supervisor.route queue, PongWorker, subscribers: number_subscribers
        [
          Thread.new { results << rpc_call.dispatch(queue, 'nap', 2) },
          Thread.new { results << rpc_call.dispatch(queue, 'nap', 1) }
        ].each(&:join)
      end

      context 'with only one subscriber' do
        let(:number_subscribers) { 1 }
        it 'will return results in the order dispatched' do
          expect(results).to eq [2, 1]
        end
      end

      context 'with multiple subscribers' do
        let(:number_subscribers) { 2 }
        it 'will return results from the faster task first' do
          expect(results).to eq [1, 2]
        end
      end

      after do
        rpc_call.close_channel
        session.close_connection
      end
    end
{'fraction_non_alphanumeric': 0.11381804339550819, 'fraction_numerical': 0.0030452988199467074, 'mean_word_length': 3.0998439937597504, 'pattern_counts': {'":': 0, '<': 9, '<?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': '7524153', 'n_tokens_mistral': 839, 'n_tokens_neox': 788, 'n_words': 183}