Class: Pgoutput::Client::SlotInspector

Inherits:
Object
  • Object
show all
Defined in:
lib/pgoutput/client/slot_inspector.rb,
sig/pgoutput/client/slot_inspector.rbs

Overview

Reads PostgreSQL replication-slot catalog state.

Inspection uses a short-lived ordinary database connection because catalog queries are separate from the long-lived replication protocol connection. Querying the whole catalog row through to_jsonb keeps the result compatible with PostgreSQL versions that expose different optional slot-health columns.

Constant Summary collapse

QUERY =

Version-tolerant catalog query for one replication slot.

Returns:

  • (String) 
<<~SQL
  SELECT to_jsonb(slot) AS slot
  FROM pg_catalog.pg_replication_slots AS slot
  WHERE slot.slot_name = $1
SQL

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(database_url:, connection_factory: nil) ⇒ SlotInspector

Returns a new instance of SlotInspector.

Parameters:

  • database_url (String) —

    PostgreSQL connection URL

  • connection_factory (#call, nil) (defaults to: nil) —

    optional connection factory for tests

  • database_url: (String)
  • connection_factory: (Object) (defaults to: nil) 


31
32
33
34
 
# File 'lib/pgoutput/client/slot_inspector.rb', line 31

def initialize(database_url:, connection_factory: nil)
  @database_url = database_url
  @connection_factory = connection_factory
end

Instance Attribute Details

#database_url ⇒ String (readonly)

Returns PostgreSQL connection URL.

Returns:

  • (String) —

    PostgreSQL connection URL



27
28
29
 
# File 'lib/pgoutput/client/slot_inspector.rb', line 27

def database_url
  @database_url
end

Instance Method Details

#fetch(slot_name) ⇒ SlotStatus?

Fetch the configured slot's current catalog state.

Parameters:

  • slot_name (String) —

    replication slot name

Returns:

  • (SlotStatus, nil) —

    snapshot, or nil when the slot is missing

Raises:



41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/pgoutput/client/slot_inspector.rb', line 41

def fetch(slot_name)
  connection = open_connection
  row = connection.exec_params(QUERY, [String(slot_name)]).first
  return nil unless row

  SlotStatus.from_catalog(parse_catalog(row.fetch("slot")))
rescue pg_error_class => e
  raise ConnectionError, e.message
ensure
  connection&.close unless connection&.finished?
end

#open_connection ⇒ Object

Returns:

  • (Object) 


55
56
57
58
59
60
 
# File 'lib/pgoutput/client/slot_inspector.rb', line 55

def open_connection
  return @connection_factory.call(database_url) if @connection_factory

  require "pg"
  PG.connect(database_url)
end

#parse_catalog(value) ⇒ Hash[(String | Symbol), untyped]

Parameters:

  • value (Object)

Returns:

  • (Hash[(String | Symbol), untyped]) 


62
63
64
65
66
 
# File 'lib/pgoutput/client/slot_inspector.rb', line 62

def parse_catalog(value)
  return value if value.is_a?(Hash)

  JSON.parse(value)
end

#pg_error_class ⇒ singleton(::PG::Error)

Returns:



68
69
70
71
 
# File 'lib/pgoutput/client/slot_inspector.rb', line 68

def pg_error_class
  require "pg"
  PG::Error
end