Class: Tapsoob::Pull

Inherits:

Operation

• Object
• Operation
• Tapsoob::Pull

Defined in:

lib/tapsoob/operation.rb

Instance Attribute Summary

Attributes inherited from Operation

#database_url, #dump_path, #opts

Class Method Summary

• .factory(db, state) ⇒ Object

Instance Method Summary

• #fetch_tables_info ⇒ Object
• #file_prefix ⇒ Object
• #pull_data ⇒ Object
• #pull_data_from_table(stream, progress) ⇒ Object
• #pull_indexes ⇒ Object
• #pull_partial_data ⇒ Object
• #pull_reset_sequences ⇒ Object
• #pull_schema ⇒ Object
• #record_count ⇒ Object
• #run ⇒ Object
• #tables ⇒ Object
• #tables_info ⇒ Object
• #to_hash ⇒ Object

Methods inherited from Operation

#apply_table_filter, #catch_errors, #completed_tables, #db, #default_chunksize, #exclude_tables, #exiting?, #format_number, #indexes_first?, #initialize, #log, #resuming?, #schema_only?, #setup_signal_trap, #skip_schema?, #store_session, #stream_state, #stream_state=, #table_filter

Constructor Details

This class inherits a constructor from Tapsoob::Operation

Class Method Details

.factory(db, state) ⇒ Object

# File 'lib/tapsoob/operation.rb', line 306

def self.factory(db, state)
  if defined?(Sequel::MySQL) && Sequel::MySQL.respond_to?(:convert_invalid_date_time=)
    Sequel::MySQL.convert_invalid_date_time = :nil
  end

  if state.has_key?(:klass)
    return eval(state[:klass]).new(db, state)
  end

  if Tapsoob::Utils.single_integer_primary_key(db, state[:table_name].to_sym)
    DataStreamKeyed.new(db, state)
  else
    DataStream.new(db, state)
  end
end

Instance Method Details

#fetch_tables_info ⇒ Object

# File 'lib/tapsoob/operation.rb', line 296

def fetch_tables_info
  tables = db.tables

  data = {}
  apply_table_filter(tables).each do |table_name|
    data[table_name] = db[table_name].count
  end
  data
end

#file_prefix ⇒ Object

# File 'lib/tapsoob/operation.rb', line 158

def file_prefix
  "pull"
end

#pull_data ⇒ Object

# File 'lib/tapsoob/operation.rb', line 194

def pull_data
  log.info "Receiving data"

  log.info "#{tables.size} tables, #{format_number(record_count)} records"

  tables.each do |table_name, count|
    progress = (opts[:progress] ? ProgressBar.new(table_name.to_s, count) : nil)
    stream   = Tapsoob::DataStream.factory(db, {
      :chunksize  => default_chunksize,
      :table_name => table_name
    }, { :debug => opts[:debug] })
    pull_data_from_table(stream, progress)
  end
end

#pull_data_from_table(stream, progress) ⇒ Object

# File 'lib/tapsoob/operation.rb', line 221

def pull_data_from_table(stream, progress)
  loop do
    if exiting?
      store_session
      exit 0
    end

    row_size = 0
    chunksize = stream.state[:chunksize]

    begin
      chunksize = Tapsoob::Utils.calculate_chunksize(chunksize) do |c|
        stream.state[:chunksize] = c.to_i
        encoded_data, row_size, elapsed_time = nil
        d1 = c.time_delta do
          encoded_data, row_size, elapsed_time = stream.fetch
        end
    
        data = nil
        d2 = c.time_delta do
          data = {
            :state        => stream.to_hash,
            :checksum     => Tapsoob::Utils.checksum(encoded_data).to_s,
            :encoded_data => encoded_data
          }
        end
    
        stream.fetch_data_from_database(data) do |rows|
          if dump_path.nil?
            puts JSON.generate(rows)
          else
            Tapsoob::Utils.export_rows(dump_path, stream.table_name, rows)
          end
        end
        log.debug "row size: #{row_size}"
        stream.error = false
        self.stream_state = stream.to_hash
        
        c.idle_secs = (d1 + d2)
        
        elapsed_time
      end
    rescue Tapsoob::CorruptedData => e
      log.info "Corrupted Data Received #{e.message}, retrying..."
      stream.error = true
      next
    end

    progress.inc(row_size) if progress
    
    break if stream.complete?
  end

  progress.finish if progress
  completed_tables << stream.table_name.to_s
  self.stream_state = {}
end

#pull_indexes ⇒ Object

# File 'lib/tapsoob/operation.rb', line 322

def pull_indexes
  log.info "Receiving indexes"

  raw_idxs = Tapsoob::Schema.indexes_individual(database_url)
  idxs     = (raw_idxs && raw_idxs.length >= 2 ? JSON.parse(raw_idxs) : {})

  apply_table_filter(idxs).each do |table, indexes|
    next unless indexes.size > 0
    progress = ProgressBar.new(table, indexes.size)
    indexes.each do |idx|
      output = Tapsoob::Utils.export_indexes(dump_path, table, idx)
      puts output if dump_path.nil? && output
      progress.inc(1)
    end
    progress.finish
  end
end

#pull_partial_data ⇒ Object

# File 'lib/tapsoob/operation.rb', line 209

def pull_partial_data
  return if stream_state == {}

  table_name = stream_state[:table_name]
  record_count = tables[table_name.to_s]
  log.info "Resuming #{table_name}, #{format_number(record_count)} records"

  progress = (opts[:progress] ? ProgressBar.new(table_name.to_s, record_count) : nil)
  stream = Tapsoob::DataStream.factory(db, stream_state)
  pull_data_from_table(stream, progress)
end

#pull_reset_sequences ⇒ Object

# File 'lib/tapsoob/operation.rb', line 340

def pull_reset_sequences
  log.info "Resetting sequences"

  output = Tapsoob::Utils.schema_bin(:reset_db_sequences, database_url)
  puts output if dump_path.nil? && output
end

#pull_schema ⇒ Object

# File 'lib/tapsoob/operation.rb', line 180

def pull_schema
  log.info "Receiving schema"

  progress = ProgressBar.new('Schema', tables.size)
  tables.each do |table_name, count|
    schema_data = Tapsoob::Schema.dump_table(database_url, table_name)
    log.debug "Table: #{table_name}\n#{schema_data}\n"
    output = Tapsoob::Utils.export_schema(dump_path, table_name, schema_data)
    puts output if dump_path.nil? && output
    progress.inc(1)
  end
  progress.finish
end

#record_count ⇒ Object

# File 'lib/tapsoob/operation.rb', line 288

def record_count
  tables_info.values.inject(:+)
end

#run ⇒ Object

# File 'lib/tapsoob/operation.rb', line 166

def run
  catch_errors do
    unless resuming?
      pull_schema if !skip_schema?
      pull_indexes if indexes_first? && !skip_schema?
    end
    setup_signal_trap
    pull_partial_data if resuming?
    pull_data unless schema_only?
    pull_indexes if !indexes_first? && !skip_schema?
    pull_reset_sequences
  end
end

#tables ⇒ Object

# File 'lib/tapsoob/operation.rb', line 279

def tables
  h = {}
  tables_info.each do |table_name, count|
    next if completed_tables.include?(table_name.to_s)
    h[table_name.to_s] = count
  end
  h
end

#tables_info ⇒ Object

# File 'lib/tapsoob/operation.rb', line 292

def tables_info
  opts[:tables_info] ||= fetch_tables_info
end

#to_hash ⇒ Object

# File 'lib/tapsoob/operation.rb', line 162

def to_hash
  super.merge(:remote_tables_info => remote_tables_info)
end