Class: Waldit::Watcher

Inherits:
Wal::StreamingWatcher
  • Object
show all
Includes:
Wal
Defined in:
lib/waldit/watcher.rb

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Watcher

Returns a new instance of Watcher.



9
10
11
12
13
 
# File 'lib/waldit/watcher.rb', line 9

def initialize(*)
  super
  initialize_connection
  @retry = false
end

Instance Method Details

#clean_attributes(table, attributes) ⇒ Object 



48
49
50
 
# File 'lib/waldit/watcher.rb', line 48

def clean_attributes(table, attributes)
  attributes.without(ignored_columns(table))
end

#ignored_columns(table) ⇒ Object 



40
41
42
 
# File 'lib/waldit/watcher.rb', line 40

def ignored_columns(table)
  (@ignored_columns_cache ||= {})[table] ||= Waldit.ignored_columns.call(table)
end

#on_transaction_events(events) ⇒ Object 



15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
 
# File 'lib/waldit/watcher.rb', line 15

def on_transaction_events(events)
  begin_event = events.next
  # The commit LSN identifies the transaction. The xid from the replication stream is the raw
  # 32-bit counter, which wraps around every ~4 billion transactions, so it cannot be used as
  # a durable identity.
  if begin_event.estimated_size < Waldit.large_transaction_threshold
    process_in_memory(begin_event.final_lsn, events)
  else
    process_streaming(begin_event.final_lsn, events)
  end
rescue PG::ConnectionBad
  raise if @retry
  initialize_connection
  @retry = true
  retry
end

#record ⇒ Object 



52
53
54
 
# File 'lib/waldit/watcher.rb', line 52

def record
  Waldit.model
end

#should_watch_table?(table) ⇒ Boolean

Returns:

  • (Boolean) 


32
33
34
 
# File 'lib/waldit/watcher.rb', line 32

def should_watch_table?(table)
  Waldit.watched_tables.call(table)
end

#store_changes(table) ⇒ Object 



44
45
46
 
# File 'lib/waldit/watcher.rb', line 44

def store_changes(table)
  (@store_changes_cache ||= {})[table] ||= Waldit.store_changes.call(table)
end

#valid_context_prefix?(prefix) ⇒ Boolean

Returns:

  • (Boolean) 


36
37
38
 
# File 'lib/waldit/watcher.rb', line 36

def valid_context_prefix?(prefix)
  prefix == Waldit.context_prefix
end