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
42
43
44
 
# File 'lib/querrel/map_reduce.rb', line 8

def map(scope, options = {}, &blk)
  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

  query_model = scope.model
  results = {}

  threads = []
  dbs.each do |db|
    threads << Thread.new do
      con_spec = retrieve_connection_spec(db, resolver)
      Thread.current[:querrel_con_spec] = con_spec
      dynamic_class = ConnectedModelFactory[query_model, con_spec]

      begin
        local_scope = dynamic_class.all.merge(scope)
        results[db] = if block_given?
          res = yield(local_scope)
          res.to_a.each(&:readonly!) if res.is_a?(ActiveRecord::Relation)
          res
        else
          local_scope.to_a.each(&:readonly!)
        end
      ensure
        dynamic_class.connection_pool.release_connection
      end
    end
  end
  threads.each(&:join)

  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 



46
47
48
 
# File 'lib/querrel/map_reduce.rb', line 46

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