Class: IndexTanked::ActiveRecordDefaults::Queue::Document

Inherits:
ActiveRecord::Base
  • Object
show all
Defined in:
lib/index-tanked/active_record_defaults/queue/document.rb

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.clear_expired_locks ⇒ Object 



31
32
33
34
35
 
# File 'lib/index-tanked/active_record_defaults/queue/document.rb', line 31

def self.clear_expired_locks
  locks_cleared = update_all(["locked_at = NULL, locked_by = NULL"],
                             ["age(clock_timestamp() at time zone 'UTC', locked_at) > interval '5 minutes'"])
  locks_cleared
end

.clear_locks_by_identifier(identifier) ⇒ Object 



25
26
27
28
29
 
# File 'lib/index-tanked/active_record_defaults/queue/document.rb', line 25

def self.clear_locks_by_identifier(identifier)
  locks_cleared = update_all(["locked_by = NULL, locked_at = NULL"],
                             ["locked_by = ?", identifier])
  locks_cleared
end

.delete_outdated_locked_records_by_identifier(identifier) ⇒ Object 



37
38
39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/index-tanked/active_record_defaults/queue/document.rb', line 37

def self.delete_outdated_locked_records_by_identifier(identifier)
  ids_to_delete = non_unique_docids_by_identifier(identifier).inject([]) do |ids, (model_name, record_id)|
    record = find_by_model_name_and_record_id_and_locked_by(model_name, record_id, identifier)
    ids << record.id unless record.newest_record_with_this_docid?
    ids
  end

  if ids_to_delete.any?
    delete_all(['id in (?)', ids_to_delete])
  else
    0
  end
end

.get_or_update_index_information(model_name) ⇒ Object 



57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/index-tanked/active_record_defaults/queue/document.rb', line 57

def self.get_or_update_index_information(model_name)
  @model_list ||= {}
  @index_list ||= {}
  if @model_list[model_name]
    @model_list[model_name]
  else
    class_companion = model_name.constantize.index_tanked
    url = class_companion.index_tank_url
    index_name = class_companion.index_name
    companion_key = update_model_list(:model_name => model_name,
                                      :url => url,
                                      :index_name => index_name)
    update_index_list(companion_key, class_companion)

  end
  @model_list[model_name]
end

.index_of_duplicate_document(document_array, document_to_index) ⇒ Object 



110
111
112
113
114
 
# File 'lib/index-tanked/active_record_defaults/queue/document.rb', line 110

def self.index_of_duplicate_document(document_array, document_to_index)
  document_array.index do |doc|
    (doc.record_id == document_to_index.record_id) && (doc.model_name == document_to_index.model_name)
  end
end

.index_tanked(companion_key) ⇒ Object 



75
76
77
 
# File 'lib/index-tanked/active_record_defaults/queue/document.rb', line 75

def self.index_tanked(companion_key)
  @index_list[companion_key]
end

.lock_records_for_batch(batch_size, identifier) ⇒ Object 



79
80
81
82
 
# File 'lib/index-tanked/active_record_defaults/queue/document.rb', line 79

def self.lock_records_for_batch(batch_size, identifier)
  update_all(["locked_by = ?, locked_at = clock_timestamp() at time zone 'UTC'", identifier],
             ["locked_by IS NULL"], :limit => batch_size)
end

.non_unique_docids_by_identifier(identifier) ⇒ Object 



51
52
53
54
55
 
# File 'lib/index-tanked/active_record_defaults/queue/document.rb', line 51

def self.non_unique_docids_by_identifier(identifier)
  find_by_sql(['SELECT model_name, record_id FROM index_tanked_documents GROUP BY model_name, record_id HAVING count(*) > 1 INTERSECT SELECT model_name, record_id FROM index_tanked_documents WHERE locked_by = ?', identifier]).map do |partial_record|
    [partial_record.model_name, partial_record.record_id]
  end
end

.partition_documents_by_companion_key(documents) ⇒ Object 



84
85
86
87
88
89
90
91
92
93
94
 
# File 'lib/index-tanked/active_record_defaults/queue/document.rb', line 84

def self.partition_documents_by_companion_key(documents)
  documents.inject({}) do |partitioned_documents, document_record|
    companion_key = get_or_update_index_information(document_record.model_name)[:companion_key]
    partitioned_documents[companion_key] ||= []
    #document may be a hash or an array of hashes due for single table inheritence ancestors
    [document_record.document].flatten.each do |document_to_index|
      partitioned_documents[companion_key] << document_to_index
    end
    partitioned_documents
  end
end

.remove_duplicate_documents(documents) ⇒ Object 



96
97
98
99
100
101
102
103
104
105
106
107
108
 
# File 'lib/index-tanked/active_record_defaults/queue/document.rb', line 96

def self.remove_duplicate_documents(documents)
  documents.inject([]) do |documents, document_record|
    duplicate_index = index_of_duplicate_document(documents, document_record)
    if duplicate_index
      if documents[duplicate_index].created_at < document_record.created_at
        documents[duplicate_index] = document_record
      end
    else
      documents << document_record
    end
    documents
  end
end

.update_index_list(companion_key, class_companion) ⇒ Object 



116
117
118
 
# File 'lib/index-tanked/active_record_defaults/queue/document.rb', line 116

def self.update_index_list(companion_key, class_companion)
  @index_list[companion_key] = class_companion unless @index_list[companion_key].present?
end

.update_model_list(options) ⇒ Object 



120
121
122
123
124
125
 
# File 'lib/index-tanked/active_record_defaults/queue/document.rb', line 120

def self.update_model_list(options)
  @model_list[options[:model_name]] = { :index_tank_url => options[:url],
                                        :index_name => options[:index_name],
                                        :companion_key => "#{options[:url]} - #{options[:index_name]}" }
  @model_list[options[:model_name]][:companion_key]
end

Instance Method Details

#document ⇒ Object 



8
9
10
 
# File 'lib/index-tanked/active_record_defaults/queue/document.rb', line 8

def document
  Marshal.load(Base64.decode64(read_attribute(:document)))
end

#document=(doc) ⇒ Object 



12
13
14
 
# File 'lib/index-tanked/active_record_defaults/queue/document.rb', line 12

def document=(doc)
  write_attribute(:document, Base64.encode64(Marshal.dump(doc)))
end

#inspect ⇒ Object 



16
17
18
19
 
# File 'lib/index-tanked/active_record_defaults/queue/document.rb', line 16

def inspect
  return super if read_attribute(:document).nil?
  super.sub(/document: \"[^\"\r\n]*\"/, %{document: #{document.inspect}})
end

#newest_record_with_this_docid? ⇒ Boolean

Returns:

  • (Boolean) 


21
22
23
 
# File 'lib/index-tanked/active_record_defaults/queue/document.rb', line 21

def newest_record_with_this_docid?
  self.class.find_by_model_name_and_record_id(self.model_name, self.record_id, :order => 'created_at DESC', :limit => 1) == self
end