Class: Taps::Pull

Inherits:

Operation

• Object
• Operation
• Taps::Pull

Defined in:

lib/taps/operation.rb

Instance Attribute Summary

Attributes inherited from Operation

#database_url, #opts, #remote_url, #session_uri

Instance Method Summary

• #fetch_remote_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
• #remote_tables_info ⇒ Object
• #run ⇒ Object
• #tables ⇒ Object
• #to_hash ⇒ Object

Methods inherited from Operation

#apply_table_filter, #close_session, #completed_tables, #compression_disabled?, #db, #default_chunksize, #exiting?, factory, #format_number, #http_headers, #indexes_first?, #initialize, #log, #resuming?, #safe_database_url, #safe_remote_url, #safe_url, #server, #session_resource, #set_session, #setup_signal_trap, #store_session, #stream_state, #stream_state=, #table_filter, #verify_server

Constructor Details

This class inherits a constructor from Taps::Operation

Instance Method Details

#fetch_remote_tables_info ⇒ Object

# File 'lib/taps/operation.rb', line 321

def fetch_remote_tables_info
	retries = 0
	max_retries = 10
	begin
		tables = JSON.load(session_resource['pull/table_names'].get(http_headers).to_s)
	rescue RestClient::Exception
		retries += 1
		retry if retries <= max_retries
		puts "Unable to fetch tables information from #{remote_url}. Please check the server log."
		exit(1)
	end

	data = {}
	apply_table_filter(tables).each do |table_name|
		retries = 0
		begin
			count = session_resource['pull/table_count'].post({:table => table_name}, http_headers).to_s.to_i
			data[table_name] = count
		rescue RestClient::Exception
			retries += 1
			retry if retries <= max_retries
			puts "Unable to fetch tables information from #{remote_url}. Please check the server log."
			exit(1)
		end
	end
	data
end

#file_prefix ⇒ Object

# File 'lib/taps/operation.rb', line 203

def file_prefix
	"pull"
end

#pull_data ⇒ Object

# File 'lib/taps/operation.rb', line 252

def pull_data
	puts "Receiving data"

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

	tables.each do |table_name, count|
		progress = ProgressBar.new(table_name.to_s, count)
		stream = Taps::DataStream.factory(db, {
			:chunksize => default_chunksize,
			:table_name => table_name
		})
		pull_data_from_table(stream, progress)
	end
end

#pull_data_from_table(stream, progress) ⇒ Object

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

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

			size = stream.fetch_remote(session_resource['pull/table'], http_headers)
			break if stream.complete?
			progress.inc(size) unless exiting?
			stream.error = false
			self.stream_state = stream.to_hash
		rescue DataStream::CorruptedData => e
			puts "Corrupted Data Received #{e.message}, retrying..."
			stream.error = true
			next
		end
	end

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

#pull_indexes ⇒ Object

# File 'lib/taps/operation.rb', line 349

def pull_indexes
	puts "Receiving indexes"

	idxs = JSON.parse(session_resource['pull/indexes'].get(http_headers).to_s)

	apply_table_filter(idxs).each do |table, indexes|
		next unless indexes.size > 0
		progress = ProgressBar.new(table, indexes.size)
		indexes.each do |idx|
			output = Taps::Utils.load_indexes(database_url, idx)
			puts output if output
			progress.inc(1)
		end
		progress.finish
	end
end

#pull_partial_data ⇒ Object

# File 'lib/taps/operation.rb', line 267

def pull_partial_data
	return if stream_state == {}

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

	progress = ProgressBar.new(table_name.to_s, record_count)
	stream = Taps::DataStream.factory(db, stream_state)
	pull_data_from_table(stream, progress)
end

#pull_reset_sequences ⇒ Object

# File 'lib/taps/operation.rb', line 366

def pull_reset_sequences
	puts "Resetting sequences"

	output = Taps::Utils.schema_bin(:reset_db_sequences, database_url)
	puts output if output
end

#pull_schema ⇒ Object

# File 'lib/taps/operation.rb', line 238

def pull_schema
	puts "Receiving schema"

	progress = ProgressBar.new('Schema', tables.size)
	tables.each do |table_name, count|
		schema_data = session_resource['pull/schema'].post({:table_name => table_name}, http_headers).to_s
		log.debug "Table: #{table_name}\n#{schema_data}\n"
		output = Taps::Utils.load_schema(database_url, schema_data)
		puts output if output
		progress.inc(1)
	end
	progress.finish
end

#record_count ⇒ Object

# File 'lib/taps/operation.rb', line 313

def record_count
	@record_count ||= remote_tables_info.values.inject(0) { |a,c| a += c }
end

#remote_tables_info ⇒ Object

# File 'lib/taps/operation.rb', line 317

def remote_tables_info
	opts[:remote_tables_info] ||= fetch_remote_tables_info
end

#run ⇒ Object

# File 'lib/taps/operation.rb', line 211

def run
	verify_server

	begin
		unless resuming?
			pull_schema
			pull_indexes if indexes_first?
		end
		setup_signal_trap
		pull_partial_data if resuming?
		pull_data
		pull_indexes unless indexes_first?
		pull_reset_sequences
		close_session
	rescue RestClient::Exception => e
		store_session
		if e.respond_to?(:response)
			puts "!!! Caught Server Exception"
			puts "HTTP CODE: #{e.http_code}"
			puts "#{e.response.to_s}"
			exit(1)
		else
			raise
		end
	end
end

#tables ⇒ Object

# File 'lib/taps/operation.rb', line 304

def tables
	h = {}
	remote_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

#to_hash ⇒ Object

# File 'lib/taps/operation.rb', line 207

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