Class: Moped::PromiscuousQueryWrapper::PromiscuousWriteOperation

Inherits:
Promiscuous::Publisher::Operation::Atomic show all
Includes:
PromiscuousHelpers
Defined in:
lib/promiscuous/publisher/operation/mongoid.rb

Instance Attribute Summary collapse

Attributes inherited from Promiscuous::Publisher::Operation::Atomic

#instance

Attributes inherited from Promiscuous::Publisher::Operation::Base

#operation

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Promiscuous::Publisher::Operation::Atomic

#acquire_op_lock, #do_database_query, #operation_payloads, #query_dependencies, #reload_instance, #yell_about_missing_instance

Methods inherited from Promiscuous::Publisher::Operation::Base

_acquire_lock, #acquire_op_lock, #dependencies_for, #dependency_for_op_lock, #ensure_op_still_locked, #execute, #explain_operation, #generate_payload, #get_new_op_lock, #increment_dependencies, lock_options, #on_rabbitmq_confirm, #operation_payloads, #payload_for, #publish_payload_in_rabbitmq_async, #publish_payload_in_redis, #query_dependencies, rabbitmq_staging_set_key, #record_timestamp, recover_locks, recover_operation_from_lock, recover_payloads_for_rabbitmq, #recovering?, register_recovery_mechanism, #release_op_lock, run_recovery_mechanisms, #trace_operation, #write_dependencies

Constructor Details

#initialize(options = {}) ⇒ PromiscuousWriteOperation

Returns a new instance of PromiscuousWriteOperation.



110
111
112
113
114
 
# File 'lib/promiscuous/publisher/operation/mongoid.rb', line 110

def initialize(options={})
  super
  @query = options[:query]
  @change = options[:change]
end

Instance Attribute Details

#change ⇒ Object

Returns the value of attribute change.



108
109
110
 
# File 'lib/promiscuous/publisher/operation/mongoid.rb', line 108

def change
  @change
end

Class Method Details

.recover_operation(collection, instance_id) ⇒ Object 



120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/promiscuous/publisher/operation/mongoid.rb', line 120

def self.recover_operation(collection, instance_id)
  # TODO We need to use the primary database. We cannot read from a secondary.
  model = Promiscuous::Publisher::Model::Mongoid.collection_mapping[collection]
  query = model.unscoped.where(:id => instance_id).query

  # We no-op the update operation instead of making it idempotent.
  # To do so, we do a dummy update on the document.
  # The original caller will fail because the lock was unlocked, so we'll
  # won't send a different message.
  new(:query => query, :change => {}).tap { |op| op.instance_eval { reload_instance } }
end

Instance Method Details

#any_published_field_changed? ⇒ Boolean

Returns:

  • (Boolean) 


185
186
187
188
189
190
191
192
193
 
# File 'lib/promiscuous/publisher/operation/mongoid.rb', line 185

def any_published_field_changed?
  return true unless @change

  # TODO maybe we should cache these things
  # TODO discover field dependencies automatically (hard)
  aliases = Hash[model.aliased_fields.map { |k,v| [v,k] }]
  attributes = fields_in_query(@change).map { |f| [aliases[f.to_s], f] }.flatten.compact.map(&:to_sym)
  (attributes & model.published_db_fields).present?
end

#execute_instrumented(query) ⇒ Object 



165
166
167
168
169
170
171
 
# File 'lib/promiscuous/publisher/operation/mongoid.rb', line 165

def execute_instrumented(query)
  # We are trying to be optimistic for the locking. We are trying to figure
  # out our dependencies with the selector upfront to avoid an extra read
  # from reload_instance.
  @instance ||= get_selector_instance unless recovering? && operation == :update
  super
end

#fetch_instance ⇒ Object 



144
145
146
147
 
# File 'lib/promiscuous/publisher/operation/mongoid.rb', line 144

def fetch_instance
  raw_instance = without_promiscuous { @query.first }
  Mongoid::Factory.from_db(model, raw_instance) if raw_instance
end

#fields_in_query(change) ⇒ Object 



173
174
175
176
177
178
179
180
181
182
183
 
# File 'lib/promiscuous/publisher/operation/mongoid.rb', line 173

def fields_in_query(change)
  # We are going to extract all the keys in any nested hashes, this will be the
  # list of fields that can potentially change during the update.
  if change.is_a?(Hash)
    fields = change.keys + change.values.map(&method(:fields_in_query)).flatten
    # The split on . is for embedded documents, we don't look further down.
    fields.map { |f| f.to_s.split('.').first}.select { |k| k.to_s =~ /^[^$]/ }.uniq
  else
    []
  end
end

#increment_version_in_document ⇒ Object 



160
161
162
163
 
# File 'lib/promiscuous/publisher/operation/mongoid.rb', line 160

def increment_version_in_document
  @change['$inc'] ||= {}
  @change['$inc'][Promiscuous::Config.version_field] = 1
end

#recover_db_operation ⇒ Object 



132
133
134
135
136
137
138
 
# File 'lib/promiscuous/publisher/operation/mongoid.rb', line 132

def recover_db_operation
  if operation == :update
    without_promiscuous { @query.update(@change) }
  else
    without_promiscuous { @query.remove }
  end
end

#recoverable_failure?(exception) ⇒ Boolean

Returns:

  • (Boolean) 


140
141
142
 
# File 'lib/promiscuous/publisher/operation/mongoid.rb', line 140

def recoverable_failure?(exception)
  exception.is_a?(Moped::Errors::ConnectionFailure)
end

#recovery_payload ⇒ Object 



116
117
118
 
# File 'lib/promiscuous/publisher/operation/mongoid.rb', line 116

def recovery_payload
  [@instance.class.promiscuous_collection_name, @instance.id]
end

#should_instrument_query? ⇒ Boolean

Returns:

  • (Boolean) 


195
196
197
 
# File 'lib/promiscuous/publisher/operation/mongoid.rb', line 195

def should_instrument_query?
  super && model && any_published_field_changed?
end

#use_id_selector(options = {}) ⇒ Object 



149
150
151
152
153
154
155
156
157
158
 
# File 'lib/promiscuous/publisher/operation/mongoid.rb', line 149

def use_id_selector(options={})
  selector = {'_id' => @instance.id}.merge(@query.selector.select { |k,v| k.to_s.include?("_id") })

  if options[:use_atomic_version_selector]
    version = @instance[Promiscuous::Config.version_field]
    selector.merge!(Promiscuous::Config.version_field => version)
  end

  @query.selector = selector
end