Synchronous Mongoid to Sunspot indexer

speedfeed_sunspot.rb

BATCH_LIMIT = 500
processed_event_count = 0

logger = Logger.new(STDOUT)

total_event_count = SearchIndexEvent.count

## had to get a little closer to the DB here due to a bug in Mongoid2
## that manifests when disabling cursor timeouts
SearchIndexEvent.collection.driver.find({}, :timeout => false) do |cursor|
  cursor.each_slice(BATCH_LIMIT) do |event_docs|
    logger.info "[ #{processed_event_count} / #{total_event_count} ]"

    processed_event_count += event_docs.size

    event_docs.reject! { |event_doc| event_doc["is_delete"] }

    entities = event_docs.collect do |event_doc|
      event_doc["record_type"].constantize.send(:find, event_doc["_id"]) rescue nil
    end.compact

    type_change_event_docs = event_docs.select { |event_doc| event_doc["previous_type"] }

    begin
      Sunspot.index(entities)
      type_change_event_docs.each do |event_doc|
        Sunspot.remove_by_id(event_doc["previous_type"].constantize, event_doc["_id"])
      end
    rescue StandardError => e # typically Errno::ECONNREFUSED, Errno::ECONNRESET but might get custom attribute nils
      logger.warn "* Caught exception while indexing: #{e.class} - #{e.message}"
      logger.warn "* Skipping to next batch without deleting this one"
      next
    end

    event_doc_ids = event_docs.collect { |event_doc| event_doc["_id"] }

    SearchIndexEvent.collection.driver.remove({ :_id => { '$in' => event_doc_ids } })
  end
end

## this cursor will eventually time out and throw an exception
# SearchIndexEvent.all.each_slice(BATCH_LIMIT) do |events|
#   logger.info "[ #{processed_event_count} / #{total_event_count} ]"

#   # incr by slice size now since we won't print til next loop
#   processed_event_count += events.size

#   # let workers handle these
#   events.reject! { |event| event.is_delete }

#   entities = events.collect do |event|
#     event.record_type.constantize.send(:find, event.id) rescue nil
#   end.compact

#   type_change_events = events.select { |event| event.has_attribute?(:previous_type) }

#   begin
#     Sunspot.index(entities)
#     type_change_events.each do |event|
#       Sunspot.remove_by_id(event.previous_type.constantize, event.id)
#     end
#   rescue Errno::ECONNREFUSED, Errno::ECONNRESET
#     next
#   end

#   events.map(&:destroy)
# end

logger.info "[ #{total_event_count} / #{total_event_count} ]"