Class: ElasticSearchFramework::Repository

Inherits:
Object
  • Object
show all
Defined in:
lib/elastic_search_framework/repository.rb

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.pool ⇒ Object 



185
186
187
 
# File 'lib/elastic_search_framework/repository.rb', line 185

def self.pool
  @pool ||= ConnectionPool.new(size: pool_size, timeout: pool_timeout) { ElasticSearchFramework::Repository.new }
end

.pool_size ⇒ Object 



189
190
191
 
# File 'lib/elastic_search_framework/repository.rb', line 189

def self.pool_size
  @pool_size ||= Integer(ENV['CONNECTION_POOL_SIZE'] || 25)
end

.pool_timeout ⇒ Object 



193
194
195
 
# File 'lib/elastic_search_framework/repository.rb', line 193

def self.pool_timeout
  @pool_timeout ||= Integer(ENV['CONNECTION_IDLE_TIMEOUT'] || 5)
end

Instance Method Details

#client ⇒ Object 



126
127
128
129
130
131
132
 
# File 'lib/elastic_search_framework/repository.rb', line 126

def client
  @client ||= Net::HTTP.new(host_uri.host, host_uri.port).tap do |c|
    c.use_ssl = host_uri.scheme == 'https'
    c.open_timeout = open_timeout
    c.read_timeout = read_timeout
  end
end

#drop(index:, id:, type: 'default', routing_key: nil) ⇒ Object 



55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/elastic_search_framework/repository.rb', line 55

def drop(index:, id:, type: 'default', routing_key: nil)
  uri_string = "#{host}/#{index.full_name}/#{type.downcase}/#{id}"
  uri_string += "?routing=#{routing_key}" if routing_key

  uri = URI(uri_string)

  request = Net::HTTP::Delete.new(uri.request_uri)

  response = with_client do |client|
    client.request(request)
  end

  if valid_response?(response.code) || Integer(response.code) == 404
    return true
  else
    raise ElasticSearchFramework::Exceptions::IndexError.new(
        "An error occurred dropping an index document. Response: #{response.body}"
    )
  end
end

#get(index:, id:, type: 'default', routing_key: nil) ⇒ Object 



30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
 
# File 'lib/elastic_search_framework/repository.rb', line 30

def get(index:, id:, type: 'default', routing_key: nil)
  uri_string = "#{host}/#{index.full_name}/#{type.downcase}/#{id}/_source"
  uri_string += "?routing=#{routing_key}" if routing_key

  uri = URI(uri_string)

  request = Net::HTTP::Get.new(uri.request_uri)

  response = with_client do |client|
    client.request(request)
  end

  if valid_response?(response.code)
    result = JSON.load(response.body)
    hash_helper.indifferent!(result)
    return result
  elsif Integer(response.code) == 404
    return nil
  else
    raise ElasticSearchFramework::Exceptions::IndexError.new(
        "An error occurred getting an index document. Response: #{response.body}"
    )
  end
end

#get_id_value(index:, entity:) ⇒ Object 



162
163
164
165
166
167
168
 
# File 'lib/elastic_search_framework/repository.rb', line 162

def get_id_value(index:, entity:)
  if entity.is_a?(Hash)
    entity[index.description[:id].to_sym] || entity[index.description[:id].to_s]
  else
    entity.instance_variable_get("@#{index.description[:id]}")
  end
end

#hash_helper ⇒ Object 



158
159
160
 
# File 'lib/elastic_search_framework/repository.rb', line 158

def hash_helper
  @hash_helper ||= HashKit::Helper.new
end

#host ⇒ Object 



150
151
152
 
# File 'lib/elastic_search_framework/repository.rb', line 150

def host
  "#{ElasticSearchFramework.host}:#{ElasticSearchFramework.port}"
end

#host_uri ⇒ Object 



154
155
156
 
# File 'lib/elastic_search_framework/repository.rb', line 154

def host_uri
  URI("#{ElasticSearchFramework.host}:#{ElasticSearchFramework.port}")
end

#idle_timeout ⇒ Object 



134
135
136
 
# File 'lib/elastic_search_framework/repository.rb', line 134

def idle_timeout
  @idle_timeout ||= Integer(ENV['CONNECTION_IDLE_TIMEOUT'] || 5)
end

#json_query(index_name:, json_query:, type: 'default', routing_key: nil) ⇒ Object 



101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
 
# File 'lib/elastic_search_framework/repository.rb', line 101

def json_query(index_name:, json_query:, type: 'default', routing_key: nil)
  uri_string = "#{host}/#{index_name}/#{type}/_search"
  uri_string += "?routing=#{routing_key}" if routing_key

  uri = URI(uri_string)

  request = Net::HTTP::Get.new(uri.request_uri)
  request.content_type = 'application/json'
  request.body = json_query

  response = with_client do |client|
    client.request(request)
  end

  if valid_response?(response.code)
    result = JSON.parse(response.body)
    return result['hits']
  else
    raise(
      ElasticSearchFramework::Exceptions::IndexError,
      "An error occurred executing an index query. Response: #{response.body}"
    )
  end
end

#open_timeout ⇒ Object 



142
143
144
 
# File 'lib/elastic_search_framework/repository.rb', line 142

def open_timeout
  @open_timeout ||= Integer(ENV['CONNECTION_OPEN_TIMEOUT'] || 1)
end

#query(index:, expression:, type: 'default', limit: 10, count: false, routing_key: nil) ⇒ Object 



76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
 
# File 'lib/elastic_search_framework/repository.rb', line 76

def query(index:, expression:, type: 'default', limit: 10, count: false, routing_key: nil)
  uri_string = "#{host}/#{index.full_name}/#{type}/_search?q=#{CGI::escape(expression)}&size=#{limit}"
  uri_string += "&routing=#{routing_key}" if routing_key

  uri = URI(uri_string)

  request = Net::HTTP::Get.new(uri.request_uri)

  response = with_client do |client|
    client.request(request)
  end

  if valid_response?(response.code)
    result = JSON.parse(response.body)
    hash_helper.indifferent!(result)
    if count
      return result[:hits][:total]
    else
      return result[:hits][:total] > 0 ? result[:hits][:hits] : []
    end
  else
    raise ElasticSearchFramework::Exceptions::IndexError.new("An error occurred executing an index query. Response: #{response.body}")
  end
end

#read_timeout ⇒ Object 



138
139
140
 
# File 'lib/elastic_search_framework/repository.rb', line 138

def read_timeout
  @read_timeout ||= Integer(ENV['CONNECTION_READ_TIMEOUT'] || 5)
end

#set(index:, entity:, type: 'default', op_type: 'index', routing_key: nil) ⇒ Object 



4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
 
# File 'lib/elastic_search_framework/repository.rb', line 4

def set(index:, entity:, type: 'default', op_type: 'index', routing_key: nil)
  uri_string = "#{host}/#{index.full_name}/#{type.downcase}/#{get_id_value(index: index, entity: entity)}?op_type=#{op_type}"
  uri_string += "&routing=#{routing_key}" if routing_key

  uri = URI(uri_string)
  hash = hash_helper.to_hash(entity)

  request = Net::HTTP::Put.new(uri.request_uri)
  request.body = JSON.dump(hash)
  request.content_type = 'application/json'

  response = with_client do |client|
    client.request(request)
  end

  if valid_response?(response.code)
    return true
  elsif op_type == 'create' && Integer(response.code) == 409
    return true
  else
    raise ElasticSearchFramework::Exceptions::IndexError.new(
        "An error occurred setting an index document. Response: #{response.body} | Code: #{response.code}"
    )
  end
end

#valid_response?(status) ⇒ Boolean

Returns:

  • (Boolean) 


146
147
148
 
# File 'lib/elastic_search_framework/repository.rb', line 146

def valid_response?(status)
  [200, 201, 202].include?(Integer(status))
end

#with_client ⇒ Object 



170
171
172
173
174
175
176
177
178
179
180
181
182
183
 
# File 'lib/elastic_search_framework/repository.rb', line 170

def with_client
  response = nil

  self.class.pool.with do |base|
    base.client.read_timeout = read_timeout
    begin
      response = yield base.client
    ensure
      base.client.read_timeout = idle_timeout
    end
  end

  response
end