Module: ElasticQueue::Persistence::ClassMethods

Defined in:
lib/elastic_queue/persistence.rb

Instance Method Summary collapse

Instance Method Details

#add_mappings ⇒ Object 



44
45
46
47
48
 
# File 'lib/elastic_queue/persistence.rb', line 44

def add_mappings
  model_classes.each do |klass|
    search_client.indices.put_mapping index: index_name, type: klass.to_s.underscore, body: klass.queue_mapping
  end
end

#bulk_index(batch_size = 10_000) ⇒ Object 



29
30
31
32
33
34
35
36
37
38
39
40
41
42
 
# File 'lib/elastic_queue/persistence.rb', line 29

def bulk_index(batch_size = 10_000)
  create_index unless index_exists?
  model_classes.each do |klass|
    # modelclass(model).includes(associations_for_index(model)).
    index_type = klass.to_s.underscore
    klass.find_in_batches(batch_size: batch_size) do |batch|
      body = []
      batch.each do |instance|
        body << { index: { _index: index_name, _id: instance.id, _type: index_type, data: instance.indexed_for_queue } }
      end
      search_client.bulk body: body
    end
  end
end

#create_index ⇒ Object 



15
16
17
18
 
# File 'lib/elastic_queue/persistence.rb', line 15

def create_index
  search_client.indices.create index: index_name
  add_mappings
end

#delete_index ⇒ Object 



20
21
22
 
# File 'lib/elastic_queue/persistence.rb', line 20

def delete_index
  search_client.indices.delete index: index_name
end

#index_exists? ⇒ Boolean

Returns:

  • (Boolean) 


6
7
8
 
# File 'lib/elastic_queue/persistence.rb', line 6

def index_exists?
  search_client.indices.exists index: index_name
end

#index_model(instance) ⇒ Object

TODO: move these to an instance?



51
52
53
 
# File 'lib/elastic_queue/persistence.rb', line 51

def index_model(instance)
  search_client.index index: index_name, id: instance.id, type: instance.class.to_s.underscore, body: instance.indexed_for_queue
end

#refresh_index ⇒ Object

not using it, but it is nice for debugging



25
26
27
 
# File 'lib/elastic_queue/persistence.rb', line 25

def refresh_index
  search_client.indices.refresh index: index_name
end

#remove_model(instance) ⇒ Object 



60
61
62
63
64
65
66
 
# File 'lib/elastic_queue/persistence.rb', line 60

def remove_model(instance)
  begin
    search_client.delete index: index_name, id: instance.id, type: instance.class.to_s.underscore
  rescue Elasticsearch::Transport::Transport::Errors::NotFound
    # just say you deleted it if it's not there!
  end
end

#reset_index ⇒ Object 



10
11
12
13
 
# File 'lib/elastic_queue/persistence.rb', line 10

def reset_index
  delete_index if index_exists?
  create_index
end

#upsert_model(instance) ⇒ Object 



55
56
57
58
 
# File 'lib/elastic_queue/persistence.rb', line 55

def upsert_model(instance)
  body = { doc: instance.indexed_for_queue, doc_as_upsert: true }
  search_client.update index: index_name, id: instance.id, type: instance.class.to_s.underscore, body: body, refresh: true
end