Class: Metacrunch::Elasticsearch::Reader

Inherits:
Object
  • Object
show all
Defined in:
lib/metacrunch/elasticsearch/reader.rb

Constant Summary collapse

DEFAULT_SCAN_SIZE = 
250
DEFAULT_SCROLL_EXPIRY_TIME = 
10.minutes

Instance Method Summary collapse

Constructor Details

#initialize(uri, body, log: false) ⇒ Reader

Returns a new instance of Reader.



12
13
14
15
16
17
18
19
20
 
# File 'lib/metacrunch/elasticsearch/reader.rb', line 12

def initialize(uri, body, log: false)
  unless uri.starts_with?("elasticsearch://")
    raise ArgumentError, "URI must be an elasticsearch URI (elasticsearch://...)"
  end

  @uri  = URI(uri)
  @body = body
  @log  = log
end

Instance Method Details

#count ⇒ Object 



47
48
49
50
51
52
53
 
# File 'lib/metacrunch/elasticsearch/reader.rb', line 47

def count
  client.count({
    body: { query: @body[:query] },
    index: @uri.index,
    type: @uri.type
  })["count"]
end

#each(&block) ⇒ Object 



22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/metacrunch/elasticsearch/reader.rb', line 22

def each(&block)
  return enum_for(__method__) unless block_given?

  search_result = client.search({
    body: @body,
    index: @uri.index,
    type: @uri.type,
    scroll: "#{DEFAULT_SCROLL_EXPIRY_TIME}s",
    search_type: "scan",
    size: DEFAULT_SCAN_SIZE
  })

  while (
    search_result = client.scroll(
      scroll: "#{DEFAULT_SCROLL_EXPIRY_TIME}s",
      scroll_id: search_result["_scroll_id"]
    ) and # don't use &&, the semantic of 'and' is important here
    search_result["hits"]["hits"].present?
  ) do
    search_result["hits"]["hits"].each do |_hit|
      yield(_hit)
    end
  end
end