Class: Wal::StreamingWatcher

Inherits:
Object
  • Object
show all
Includes:
Watcher
Defined in:
lib/wal/streaming_watcher.rb

Overview

A watcher that streams all the events of each WAL transaction on a separate thread.

Useful to improve the throughput, as it will allow you to process events while fetching for more in parallel.

Example:

Watcher that persists all delete events as it arrives using a single database transaction, and without waiting for the full WAL log transaction to be finished.

class RegisterDeletesWalWatcher < Wal::StreamingWalWatcher
  def on_transaction_events(events)
    DeletedApplicationRecord.transaction do
      events
        .lazy
        .filter { |event| event.is_a? DeleteEvent }
        .each { |event| DeletedApplicationRecord.create_from_event(event) }
    end
  end
end

Instance Method Summary collapse

Methods included from Watcher

#should_watch_table?, #valid_context_prefix?

Instance Method Details

#on_event(event) ⇒ Object 



32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/wal/streaming_watcher.rb', line 32

def on_event(event)
  case event
  when BeginTransactionEvent
    @event_queue = SizedQueue.new(queue_size(event))
    event_stream = Enumerator.new do |y|
      while (item = @event_queue.pop)
        y << item
        break if item.is_a?(CommitTransactionEvent)
      end
    end
    ensure_worker
    @transaction_queue << event_stream
    @event_queue << event

  when CommitTransactionEvent
    @event_queue << event
    result = @completion_queue.pop
    @event_queue.clear
    raise result if result.is_a? Exception

  else
    @event_queue << event
  end
end

#on_transaction_events(events) ⇒ Object 



26
 
# File 'lib/wal/streaming_watcher.rb', line 26

def on_transaction_events(events); end

#queue_size(event) ⇒ Object 



28
29
30
 
# File 'lib/wal/streaming_watcher.rb', line 28

def queue_size(event)
  5_000
end