Module: Querrel::MapReduce

Included in:
Querreller
Defined in:
lib/querrel/map_reduce.rb

Instance Method Summary collapse

Instance Method Details

#map(scope, options = {}, &blk) ⇒ Object 



8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/querrel/map_reduce.rb', line 8

def map(scope, options = {}, &blk)
  options = @options.merge(options)
  if options.key?(:on)
    resolver = ConnectionResolver.new(options[:on], !!options[:db_names])
    dbs = resolver.configurations.keys
  else
    resolver = @connection_resolver
    dbs = @connection_resolver.configurations.keys
  end

  results = {}
  results_semaphore = Mutex.new

  pool = StaticPool.new(options[:threads] || 20)
  dbs.each do |db|
    pool.enqueue do
      ActiveRecord::Base.connection_handler = ActiveRecord::ConnectionAdapters::ConnectionHandler.new
      con_config = retrieve_connection_config(db, resolver)
      ActiveRecord::Base.establish_connection(con_config)

      begin
        local_results = block_given? ? yield(scope, db) : scope
        local_results.to_a.each(&:readonly!) if local_results.is_a?(ActiveRecord::Relation)

        results_semaphore.synchronize { results[db] = local_results }
      ensure
        ActiveRecord::Base.connection_handler.clear_all_connections!
      end
    end
  end
  pool.do_your_thang!

  results
end

#query(scope, options = {}, &blk) ⇒ Object 



3
4
5
6
 
# File 'lib/querrel/map_reduce.rb', line 3

def query(scope, options = {}, &blk)
  buckets = map(scope, options, &blk)
  reduce(buckets)
end

#reduce(buckets) ⇒ Object 



43
44
45
 
# File 'lib/querrel/map_reduce.rb', line 43

def reduce(buckets)
  buckets.flat_map{ |db, results| results }
end