Class: PgEventstore::ReplicaSubscriptionHandler

Inherits:
Object
  • Object
show all
Defined in:
lib/pg_eventstore/subscriptions/replica_subscription_handler.rb,
sig/pg_eventstore/subscriptions/replica_subscription_handler.rbs

Instance Method Summary collapse

Constructor Details

#initialize(config_name, replica_config_name) ⇒ ReplicaSubscriptionHandler

Returns a new instance of ReplicaSubscriptionHandler.

Parameters:

  • (Symbol)
  • (Symbol) 


5
6
7
8
 
# File 'lib/pg_eventstore/subscriptions/replica_subscription_handler.rb', line 5

def initialize(config_name, replica_config_name)
  @replica_config_name = replica_config_name
  @config_name = config_name
end

Instance Method Details

#call(indexes) ⇒ void

This method returns an undefined value.

Parameters:



12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
 
# File 'lib/pg_eventstore/subscriptions/replica_subscription_handler.rb', line 12

def call(indexes)
  indexes = reject_already_processed(indexes)
  runner = AsyncRunner.new
  objects_to_migrate = {}
  runner.async do
    objects_to_migrate[:raw_events] = source_replica_queries.load_events(indexes)
  end
  runner.async do
    objects_to_migrate[:markers_index] = source_replica_queries.load_event_markers_index(indexes)
    objects_to_migrate[:markers] = source_replica_queries.load_markers(objects_to_migrate[:markers_index])
  end
  runner.async do
    objects_to_migrate[:events_index] = source_replica_queries.load_events_global_index(indexes)
    objects_to_migrate[:streams_index] = source_replica_queries.load_streams_global_index(
      objects_to_migrate[:events_index]
    )
    objects_to_migrate[:partitions] = source_replica_queries.load_partitions(objects_to_migrate[:events_index])
  end
  runner.run

  sql = []
  sql << records_to_sql(
    QueryBuilders::PartitionsFiltering::TABLE_NAME,
    Partition.options.map(&:name),
    objects_to_migrate[:partitions].map(&:options_hash),
    on_conflict: 'on conflict do nothing'
  )
  sql << records_to_sql(
    Event::PRIMARY_TABLE_NAME,
    RawEvent.options.map(&:name),
    objects_to_migrate[:raw_events].map(&:options_hash)
  )
  sql << records_to_sql(
    QueryBuilders::EventsGlobalIndexFiltering::PRIMARY_TABLE_NAME,
    EventGlobalIndex.options.map(&:name),
    objects_to_migrate[:events_index].map(&:options_hash)
  )
  sql << records_to_sql(
    QueryBuilders::StreamsGlobalIndexFiltering::PRIMARY_TABLE_NAME,
    StreamGlobalIndex.options.map(&:name),
    objects_to_migrate[:streams_index].map(&:options_hash),
    on_conflict: 'on conflict (id) do update set stream_revision = EXCLUDED.stream_revision'
  )
  sql << records_to_sql(
    QueryBuilders::EventSubscriptionPositionsFiltering::PRIMARY_TABLE_NAME,
    %i[global_position subscription_position],
    indexes.map(&:to_subscription_position_attrs)
  )
  if objects_to_migrate[:markers_index].any?
    sql << records_to_sql(
      QueryBuilders::EventMarkersIndexFiltering::PRIMARY_TABLE_NAME,
      EventMarkerIndex.options.map(&:name),
      objects_to_migrate[:markers_index].map(&:options_hash)
    )
  end
  if objects_to_migrate[:markers].any?
    sql << records_to_sql(
      QueryBuilders::EventMarkersFiltering::PRIMARY_TABLE_NAME,
      EventMarker.options.map(&:name),
      objects_to_migrate[:markers].map(&:options_hash),
      on_conflict: 'on conflict do nothing'
    )
  end
  sql = sql.join("\n")
  replica_transaction_queries.transaction(:read_committed) do
    replica_connection.with do |conn|
      conn.exec(sql)
    end
  end
end

#connection ⇒ PgEventstore::Connection

Returns:



116
117
118
 
# File 'lib/pg_eventstore/subscriptions/replica_subscription_handler.rb', line 116

def connection
  PgEventstore.connection(@config_name)
end

#destination_replica_queries ⇒ PgEventstore::ReplicaQueries

Returns:

  • (PgEventstore::ReplicaQueries) 


136
137
138
 
# File 'lib/pg_eventstore/subscriptions/replica_subscription_handler.rb', line 136

def destination_replica_queries
  ReplicaQueries.new(replica_connection, QueryStrategy::Foreground.new(replica_connection))
end

#records_to_sql(table_name, attribute_names, attributes_collection, on_conflict: nil) ⇒ String

Parameters:

  • table_name (String)
  • attribute_names (Array<Symbol>)
  • attributes_collection (Array<Hash<Symbol, Object>>)
  • on_conflict (String, nil) (defaults to: nil)

Returns:

  • (String) 


101
102
103
104
105
106
107
108
109
110
111
112
113
 
# File 'lib/pg_eventstore/subscriptions/replica_subscription_handler.rb', line 101

def records_to_sql(table_name, attribute_names, attributes_collection, on_conflict: nil)
  sql_values = attributes_collection.map do |attributes|
    values = attribute_names.map do |attribute_name|
      connection.with do |conn|
        conn.prepared_value(attributes[attribute_name])
      end
    end
    "(#{values.join(', ')})"
  end
  sql_values = sql_values.join(', ')
  attributes = attribute_names.join(', ')
  "insert into #{table_name} (#{attributes}) values #{sql_values} #{on_conflict if on_conflict};"
end

#reject_already_processed(indexes) ⇒ Array<PgEventstore::EventGlobalIndex::SubscriptionRepr>

Parameters:

Returns:



87
88
89
90
91
92
93
94
 
# File 'lib/pg_eventstore/subscriptions/replica_subscription_handler.rb', line 87

def reject_already_processed(indexes)
  existing_positions = destination_replica_queries.load_subscription_positions(indexes)
  return indexes if existing_positions.empty?

  indexes.reject do |index|
    existing_positions.include?(index.subscription_position)
  end
end

#replica_connection ⇒ PgEventstore::Connection

Returns:



121
122
123
 
# File 'lib/pg_eventstore/subscriptions/replica_subscription_handler.rb', line 121

def replica_connection
  PgEventstore.connection(@replica_config_name)
end

#replica_transaction_queries ⇒ PgEventstore::TransactionQueries

Returns:

  • (PgEventstore::TransactionQueries) 


126
127
128
 
# File 'lib/pg_eventstore/subscriptions/replica_subscription_handler.rb', line 126

def replica_transaction_queries
  TransactionQueries.new(replica_connection)
end

#source_replica_queries ⇒ PgEventstore::ReplicaQueries

Returns:

  • (PgEventstore::ReplicaQueries) 


131
132
133
 
# File 'lib/pg_eventstore/subscriptions/replica_subscription_handler.rb', line 131

def source_replica_queries
  ReplicaQueries.new(connection, QueryStrategy::Async.new(connection))
end