Module: Changebase::Replication::ActiveRecord::PostgreSQLAdapter

Defined in:
lib/changebase/replication/active_record.rb

Constant Summary collapse

CHANGEBASE_COMMENT_REGEX = 
%r{(?:--.*\n)|/\*(?:[^*]|\*[^/])*\*/}m
CHANGEBASE_READ_QUERY = 
changebase_build_read_query_regexp(
  :close, :declare, :fetch, :move, :set, :show
)

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.changebase_build_read_query_regexp(*parts) ⇒ Object

:nodoc:



78
79
80
81
82
 
# File 'lib/changebase/replication/active_record.rb', line 78

def self.changebase_build_read_query_regexp(*parts) # :nodoc:
  parts += [:begin, :commit, :explain, :release, :rollback, :savepoint, :select, :with]
  parts = parts.map { |part| /#{part}/i }
  /\A(?:[(\s]|#{CHANGEBASE_COMMENT_REGEX})*#{Regexp.union(*parts)}/
end

Instance Method Details

#commit_db_transaction ⇒ Object 



58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/changebase/replication/active_record.rb', line 58

def commit_db_transaction
  if !@without_changebase && @changebase_metadata && !@changebase_metadata.empty?
    sql = ::ActiveRecord::Base.send(:replace_named_bind_variables, <<~SQL, {version: 1, metadata: ActiveSupport::JSON.encode(@changebase_metadata)})
      INSERT INTO #{quote_table_name(Changebase.metadata_table)} ( version, data )
      VALUES ( :version, :metadata )
      ON CONFLICT ( version )
      DO UPDATE SET version = :version, data = :metadata;
    SQL

    log(sql, "CHANGEBASE") do
      ActiveSupport::Dependencies.interlock.permit_concurrent_loads do
        @connection.async_exec(sql)
      end
    end
  end
  super
end

#create_database(name, options = {}) ⇒ Object 



26
27
28
 
# File 'lib/changebase/replication/active_record.rb', line 26

def create_database(name, options = {})
  without_changebase { super }
end

#drop_database(name) ⇒ Object

:nodoc:



18
19
20
 
# File 'lib/changebase/replication/active_record.rb', line 18

def drop_database(name) # :nodoc:
  without_changebase { super }
end

#drop_table(table_name, **options) ⇒ Object

:nodoc:



22
23
24
 
# File 'lib/changebase/replication/active_record.rb', line 22

def drop_table(table_name, **options) # :nodoc:
  without_changebase { super }
end

#exec_delete(sql, name = nil, binds = []) ⇒ Object

:nodoc:



50
51
52
53
54
55
56
 
# File 'lib/changebase/replication/active_record.rb', line 50

def exec_delete(sql, name = nil, binds = []) # :nodoc:
  if !@without_changebase && !current_transaction.open? && write_query?(sql)
    transaction { super }
  else
    super
  end
end

#exec_query(sql, name = "SQL", binds = [], prepare: false) ⇒ Object 



42
43
44
45
46
47
48
 
# File 'lib/changebase/replication/active_record.rb', line 42

def exec_query(sql, name = "SQL", binds = [], prepare: false)
  if !@without_changebase && !current_transaction.open? && write_query?(sql)
    transaction { super }
  else
    super
  end
end

#execute(sql, name = nil) ⇒ Object 



34
35
36
37
38
39
40
 
# File 'lib/changebase/replication/active_record.rb', line 34

def execute(sql, name = nil)
  if !@without_changebase && !current_transaction.open? && write_query?(sql)
    transaction { super }
  else
    super
  end
end

#initialize(*args, **margs) ⇒ Object 



5
6
7
8
9
 
# File 'lib/changebase/replication/active_record.rb', line 5

def initialize(*args, **margs)
  @without_changebase = false
  @changebase_metadata = nil
  super
end

#recreate_database(name, options = {}) ⇒ Object

:nodoc:



30
31
32
 
# File 'lib/changebase/replication/active_record.rb', line 30

def recreate_database(name, options = {}) # :nodoc:
  without_changebase { super }
end

#without_changebase ⇒ Object 



11
12
13
14
15
16
 
# File 'lib/changebase/replication/active_record.rb', line 11

def without_changebase
  @without_changebase = true
  yield
ensure
  @without_changebase = false
end

#write_query?(sql) ⇒ Boolean

Returns:

  • (Boolean) 


87
88
89
90
91
 
# File 'lib/changebase/replication/active_record.rb', line 87

def write_query?(sql)
  !CHANGEBASE_READ_QUERY.match?(sql)
rescue ArgumentError # Invalid encoding
  !CHANGEBASE_READ_QUERY.match?(sql.b)
end