BulkProcessor

gem 'bulk-processor'

# Gemfile
gem 'activejob'
gem 'bulk-processor'
gem 'resque'

BulkProcessor.back_end = :active_job
BulkProcessor.queue_adapter = <adapter>

# Gemfile
gem 'dynosaur'
gem 'resque'

BulkProcessor.back_end = :dynosaur
BulkProcessor.heroku.api_key = 'my-heroku-api-key'
BulkProcessor.heroku.app_name = 'my-heroku-app-name'

# Rakefile
require 'bulk_processor/back_end/dynosaur/tasks'

# If you do not already have an :enivronment rake task, create a no-op one as
# Dynosaur tasks depend on it.
task :environment
end

BulkProcessor.temp_directory = '/tmp'
BulkProcessor.aws.access_key_id = 'my-aws-access-key'
BulkProcessor.aws.secret_access_key = 'my-aws-secret'
BulkProcessor.aws.bucket = 'my-s3-bucket'

class PetCSVProcessor
  # @return [Array<String>] column headers that must be present
  def self.required_columns
    ['species', 'name', 'age']
  end

  # @return [Array<String>] column headers that may be present. If a column
  #   header is present that is not in 'required_columns' or 'optional_columns',
  #   the file will be considered invalid and no rows will be processed.
  def self.optional_columns
    ['favorite_toy', 'talents']
  end

  def initialize(csv, payload:)
    # Assign instance variables and do any other setup
  end

  def start
    # Process the CSV
  end
end

require 'bulk_processor/csv_processor'

class PetCSVProcessor < BulkProcessor::CSVProcessor
  # Note: this must be overridden in a subclass
  #
  # @return [Array<String>] column headers that must be present
  def self.required_columns
    ['species', 'name', 'age']
  end

  # @return [Array<String>] column headers that may be present. If a column
  #   header is present that is not in 'required_columns' or 'optional_columns',
  #   the file will be considered invalid and no rows will be processed.
  def self.optional_columns
    ['favorite_toy', 'talents']
  end

  # Note: this must be overridden in a subclass
  #
  # @return [RowProcessor] a class that implements the RowProcessor role
  def self.row_processor_class
    PetRowProcessor
  end

  # @return [PreProcessor] a class that implements the PreProcessor role
  def self.pre_processor_class
    PetPreProcessor
  end

  # @return [PostProcessor] a class that implements the PostProcessor role
  def self.post_processor_class
    PetPostProcessor
  end

  # @return [Handler] a class that implements the Handler role
  def self.handler_class
    PetHandler
  end
end

class PetRowProcessor < BulkProcessor::CSVProcessor::RowProcessor
  # Process the row, e.g. create a new record in the DB, send an email, etc
  def process!
    pet = Pet.new(row)
    if pet.save
      self.successful = true
    else
      messages.concat(pet.errors.full_messages)
    end
  end

  # Setting these allow us to identify error messages by these key/values for
  # a row, rather than using the row number
  def primary_keys
    ['species', 'name']
  end
end

class PetPostProcessor
  attr_reader :results

  def initialize(row_processors)
    # Assign instance variables and do any other setup
  end

  def start
    cat_count = 0
    @results = []
    row_processors.each do |row_processor|
      cat_count += 1 if row_processor.cat?
    end

    if cat_count > 2
      @results << BulkProcessor::CSVProcessor::Result.new(messages: ['Too many cats!'],
                                                          successful: false)
    end
  end
end

class PetPreProcessor
  attr_reader :results

  def initialize(row_processors)
    # Assign instance variables and do any other setup
  end

  def start
    @results = []
    row_count = row_processors.length
    if row_count > 2
      @results << BulkProcessor::CSVProcessor::Result.new(messages: ['Too many rows!'], successful: false)
    end
  end

  def self.fail_process_if_failed
    # a class method that's used to indicate if the csv process needs to be continue with PreProcessor failed
    true
  end
end

class PetHandler
  # @param payload [Hash] the payload passed into 'BulkProcessor.process', can
  #   be used to pass metadata around, e.g. the email address to send a
  #   completion report to
  # @param results [Array<BulkProcessor::CSVProcessor::RowProcessor>] results
  #   for processing the rows (there will be one pre row in the CSV plus zero
  #   or more from post-processing)
  def initialize(payload:, results:)
    # Assign instance variables and do any other setup
  end

  # Notify the owner that their pets were processed
  def complete!
    OwnerMailer.completed(results, payload)
  end

  # Notify the owner that processing failed
  #
  # @param fatal_error [StandardError] if nil, then all rows were processed,
  #   else the error that was raise is passed in here
  def fail!(fatal_error)
    OwnerMailer.failed(fatal_error, payload)
  end
end

processor = BulkProcessor.new(
              key: file_name,
              stream: file_stream,
              processor_class: PetCSVProcessor,
              payload: { recipient: current_user.email }
            )
if processor.start
  # The job has been enqueued, go get a coffee and wait
else
  # Something went wrong, alert the file uploader
  handle_invalid_file(processor.errors)
end

processor = BulkProcessor.new(
              key: file_name,
              stream: file_stream,
              processor_class: PetCSVProcessor,
              payload: { recipient: current_user.email }
            )
if processor.start(5)
  # Split the main CSV into 5 smaller files and process in parallel.
else
  # Something went wrong, alert the file uploader
  handle_invalid_file(processor.errors)
end

class PetCSVProcessor
  def self.boundary_column
    'pet_id'
  end
  ...
end

class PetCSVProcessor
  def self.handler_class
    PetHandler
  end
  ...
end