Class: Tapsoob::Push
Instance Attribute Summary
Attributes inherited from Operation
#database_url, #dump_path, #opts
Instance Method Summary
collapse
Methods inherited from Operation
#apply_table_filter, #catch_errors, #completed_tables, #db, #default_chunksize, #exclude_tables, #exiting?, factory, #format_number, #indexes_first?, #initialize, #log, #resuming?, #setup_signal_trap, #skip_schema?, #store_session, #stream_state, #stream_state=, #table_filter
Instance Method Details
#fetch_local_tables_info ⇒ Object
477
478
479
480
481
482
483
484
485
486
487
488
489
|
# File 'lib/tapsoob/operation.rb', line 477
def fetch_local_tables_info
tables_with_counts = {}
tbls = Dir.glob(File.join(dump_path, "schemas", "*")).map { |path| File.basename(path, ".rb") }
tbls.each do |table|
if File.exists?(File.join(dump_path, "data", "#{table}.json"))
data = JSON.parse(File.read(File.join(dump_path, "data", "#{table}.json")))
tables_with_counts[table] = data.size
else
tables_with_counts[table] = 0
end
end
apply_table_filter(tables_with_counts)
end
|
#file_prefix ⇒ Object
316
317
318
|
# File 'lib/tapsoob/operation.rb', line 316
def file_prefix
"push"
end
|
#local_tables_info ⇒ Object
460
461
462
|
# File 'lib/tapsoob/operation.rb', line 460
def local_tables_info
opts[:local_tables_info] ||= fetch_local_tables_info
end
|
#push_data ⇒ Object
389
390
391
392
393
394
395
396
397
398
399
400
401
402
|
# File 'lib/tapsoob/operation.rb', line 389
def push_data
log.info "Sending data"
log.info "#{tables.size} tables, #{format_number(record_count)} records"
tables.each do |table_name, count|
next unless File.exists?(File.join(dump_path, "data", "#{table_name}.json"))
stream = Tapsoob::DataStream.factory(db,
:table_name => table_name,
:chunksize => default_chunksize)
progress = ProgressBar.new(table_name.to_s, count)
push_data_from_file(stream, progress)
end
end
|
#push_data_from_file(stream, progress) ⇒ Object
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
|
# File 'lib/tapsoob/operation.rb', line 404
def push_data_from_file(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({ :type => "file", :source => dump_path })
end
break if stream.complete?
data = nil
d2 = c.time_delta do
data = {
:state => stream.to_hash,
:checksum => Tapsoob::Utils.checksum(encoded_data).to_s
}
end
size = stream.fetch_data_in_database({ :encoded_data => encoded_data, :checksum => data[:checksum] })
self.stream_state = stream.to_hash
c.idle_secs = (d1 + d2)
elapsed_time
end
rescue Tapsoob::CorruptedData => e
next
rescue Tapsoob::DuplicatePrimaryKeyError => e
stream.verify_stream
stream = JSON.generate({ :state => stream.to_hash })
next
end
stream.state[:chunksize] = chunksize
progress.inc(row_size)
stream.increment(row_size)
break if stream.complete?
end
progress.finish
completed_tables << stream.table_name.to_s
self.stream_state = {}
end
|
#push_indexes ⇒ Object
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
|
# File 'lib/tapsoob/operation.rb', line 338
def push_indexes
idxs = {}
table_idxs = Dir.glob(File.join(dump_path, "indexes", "*.json")).map { |path| File.basename(path, '.json') }
table_idxs.each do |table_idx|
idxs[table_idx] = JSON.parse(File.read(File.join(dump_path, "indexes", "#{table_idx}.json")))
end
return unless idxs.size > 0
log.info "Sending indexes"
apply_table_filter(idxs).each do |table, indexes|
next unless indexes.size > 0
progress = ProgressBar.new(table, indexes.size)
indexes.each do |idx|
Tapsoob::Utils.load_indexes(database_url, idx)
progress.inc(1)
end
progress.finish
end
end
|
#push_partial_data ⇒ Object
378
379
380
381
382
383
384
385
386
387
|
# File 'lib/tapsoob/operation.rb', line 378
def push_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 = ProgressBar.new(table_name.to_s, record_count)
stream = Tapsoob::DataStream.factory(db, stream_state)
push_data_from_file(stream, progress)
end
|
#push_reset_sequences ⇒ Object
372
373
374
375
376
|
# File 'lib/tapsoob/operation.rb', line 372
def push_reset_sequences
log.info "Resetting sequences"
Tapsoob::Utils.schema_bin(:reset_db_sequences, database_url)
end
|
#push_schema ⇒ Object
360
361
362
363
364
365
366
367
368
369
370
|
# File 'lib/tapsoob/operation.rb', line 360
def push_schema
log.info "Sending schema"
progress = ProgressBar.new('Schema', tables.size)
tables.each do |table, count|
log.debug "Loading '#{table}' schema\n"
Tapsoob::Utils.load_schema(dump_path, database_url, table)
progress.inc(1)
end
progress.finish
end
|
#record_count ⇒ Object
473
474
475
|
# File 'lib/tapsoob/operation.rb', line 473
def record_count
@record_count ||= local_tables_info.values.inject(0) { |a,c| a += c }
end
|
#run ⇒ Object
324
325
326
327
328
329
330
331
332
333
334
335
336
|
# File 'lib/tapsoob/operation.rb', line 324
def run
catch_errors do
unless resuming?
push_schema if !skip_schema?
push_indexes if indexes_first? && !skip_schema?
end
setup_signal_trap
push_partial_data if resuming?
push_data
push_indexes if !indexes_first? && !skip_schema?
push_reset_sequences
end
end
|
#tables ⇒ Object
464
465
466
467
468
469
470
471
|
# File 'lib/tapsoob/operation.rb', line 464
def tables
h = {}
local_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
320
321
322
|
# File 'lib/tapsoob/operation.rb', line 320
def to_hash
super.merge(:local_tables_info => local_tables_info)
end
|