Class: RR::ReplicationHelper

Inherits:
Object
  • Object
show all
Includes:
LogHelper
Defined in:
lib/rubyrep/replication_helper.rb

Overview

Provides helper functionality for replicators. The methods exposed by this class are intended to provide a stable interface for third party replicators.

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from LogHelper

#fit_description_columns

Constructor Details

#initialize(replication_run) ⇒ ReplicationHelper

Creates a new SyncHelper for the given TableSync instance.



133
134
135
136
137
138
139
140
 
# File 'lib/rubyrep/replication_helper.rb', line 133

def initialize(replication_run)
  self.replication_run = replication_run

  # Creates the committer. Important as it gives the committer the
  # opportunity to start transactions
  committer_class = Committers::committers[options[:committer]]
  @committer = committer_class.new(session)
end

Instance Attribute Details

#replication_run ⇒ Object

The current ReplicationRun instance



11
12
13
 
# File 'lib/rubyrep/replication_helper.rb', line 11

def replication_run
  @replication_run
end

Instance Method Details

#corresponding_table(db_arm, table) ⇒ Object

Delegates to Session#corresponding_table



30
 
# File 'lib/rubyrep/replication_helper.rb', line 30

def corresponding_table(db_arm, table); session.corresponding_table(db_arm, table); end

#delete_record(database, table, values) ⇒ Object

Delegates to Committers::BufferedCommitter#delete_record



49
50
51
 
# File 'lib/rubyrep/replication_helper.rb', line 49

def delete_record(database, table, values)
  committer.delete_record(database, table, values)
end

#finalize(success = true) ⇒ Object

Asks the committer (if it exists) to finalize any open transactions success should be true if there were no problems, false otherwise.



76
77
78
 
# File 'lib/rubyrep/replication_helper.rb', line 76

def finalize(success = true)
  committer.finalize(success)
end

#insert_record(database, table, values) ⇒ Object

Delegates to Committers::BufferedCommitter#insert_record



39
40
41
 
# File 'lib/rubyrep/replication_helper.rb', line 39

def insert_record(database, table, values)
  committer.insert_record(database, table, values)
end

#load_record(database, table, key) ⇒ Object

Loads the specified record. Returns an according column_name => value hash. Parameters:

  • database: either :left or :right

  • table: name of the table

  • key: A column_name => value hash for all primary key columns.



58
59
60
61
62
63
64
65
66
67
68
 
# File 'lib/rubyrep/replication_helper.rb', line 58

def load_record(database, table, key)
  cursor = session.send(database).select_cursor(
    :table => table,
    :row_keys => [key],
    :type_cast => true
  )
  row = nil
  row = cursor.next_row if cursor.next?
  cursor.clear
  row
end

#log_replication_outcome(diff, outcome, details = nil) ⇒ Object

Logs the outcome of a replication into the replication log table.

  • diff: the replicated ReplicationDifference

  • outcome: string summarizing the outcome of the replication

  • details: string with further details regarding the replication



104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/rubyrep/replication_helper.rb', line 104

def log_replication_outcome(diff, outcome, details = nil)
  table = diff.changes[:left].table
  key = diff.changes[:left].key
  if key.size == 1
    key = key.values[0]
  else
    key_parts = session.left.primary_key_names(table).map do |column_name|
      %Q("#{column_name}"=>#{key[column_name].to_s.inspect})
    end
    key = key_parts.join(', ')
  end
  rep_outcome, rep_details = fit_description_columns(outcome, details)
  diff_dump = diff.to_yaml[0...ReplicationInitializer::DIFF_DUMP_SIZE]
  
  session.left.insert_record "#{options[:rep_prefix]}_logged_events", {
    :activity => 'replication',
    :change_table => table,
    :diff_type => diff.type.to_s,
    :change_key => key,
    :left_change_type => (diff.changes[:left] ? diff.changes[:left].type.to_s : nil),
    :right_change_type => (diff.changes[:right] ? diff.changes[:right].type.to_s : nil),
    :description => rep_outcome,
    :long_description => rep_details,
    :event_time => Time.now,
    :diff_dump => diff_dump
  }
end

#new_transaction? ⇒ Boolean

Returns true if a new transaction was started since the last insert / update / delete.

Returns:

  • (Boolean) 


34
35
36
 
# File 'lib/rubyrep/replication_helper.rb', line 34

def new_transaction?
  committer.new_transaction?
end

#options ⇒ Object

Current options



17
 
# File 'lib/rubyrep/replication_helper.rb', line 17

def options; @options ||= session.configuration.options; end

#options_for_table(table) ⇒ Object

Returns the options for the specified table name.

  • table: name of the table (left database version)



21
22
23
24
25
26
27
 
# File 'lib/rubyrep/replication_helper.rb', line 21

def options_for_table(table)
  @options_for_table ||= {}
  unless @options_for_table.include? table
    @options_for_table[table] = session.configuration.options_for_table(table)
  end
  @options_for_table[table]
end

#session ⇒ Object

The active Session



14
 
# File 'lib/rubyrep/replication_helper.rb', line 14

def session; replication_run.session; end

#type_cast(table, row) ⇒ Object

Converts the row values into their proper types as per table definition.

  • table: name of the table after whose columns is type-casted.

  • row: A column_name => value hash of the row

Returns a copy of the column_name => value hash (with type-casted values).



84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/rubyrep/replication_helper.rb', line 84

def type_cast(table, row)
  @table_columns ||= {}
  unless @table_columns.include?(table)
    column_array = session.left.columns(table)
    column_hash = {}
    column_array.each {|column| column_hash[column.name] = column}
    @table_columns[table] = column_hash
  end
  columns = @table_columns[table]
  type_casted_row = {}
  row.each_pair do |column_name, value|
    type_casted_row[column_name] = columns[column_name].type_cast(value)
  end
  type_casted_row
end

#update_record(database, table, values, old_key = nil) ⇒ Object

Delegates to Committers::BufferedCommitter#update_record



44
45
46
 
# File 'lib/rubyrep/replication_helper.rb', line 44

def update_record(database, table, values, old_key = nil)
  committer.update_record(database, table, values, old_key)
end