Class: Pgoutput::Client::Configuration

Inherits:
Object
  • Object
show all
Defined in:
lib/pgoutput/client/configuration.rb

Overview

Immutable configuration for a PostgreSQL logical replication stream.

A configuration describes how ‘pgoutput-client` should connect to PostgreSQL and how it should request logical replication from the server. It deliberately contains transport-level settings only; parsing pgoutput records and decoding PostgreSQL values belong to downstream layers.

The object freezes itself and its string/array attributes during initialization so it can be safely shared by transport components without defensive copying.

Examples:

Minimal configuration

config = Pgoutput::Client::Configuration.new(
  database_url: "postgres://localhost/app",
  slot_name: "cdc_slot",
  publication_names: "app_publication"
)

Start from a known LSN and request binary values from pgoutput

config = Pgoutput::Client::Configuration.new(
  database_url: ENV.fetch("DATABASE_URL"),
  slot_name: "cdc_slot",
  publication_names: %w[app_publication],
  start_lsn: "0/16B6C50",
  binary: true
)

Constant Summary collapse

DEFAULT_PLUGIN =

Default logical decoding output plugin.

Returns:

  • (String) 
"pgoutput"
DEFAULT_PROTO_VERSION =

Default pgoutput protocol version.

Returns:

  • (Integer) 
1
DEFAULT_FEEDBACK_INTERVAL =

Default interval, in seconds, between standby status feedback messages.

Returns:

  • (Float) 
10.0

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(database_url:, slot_name:, publication_names:, start_lsn: nil, plugin: DEFAULT_PLUGIN, proto_version: DEFAULT_PROTO_VERSION, binary: false, messages: false, auto_create_slot: false, temporary_slot: false, feedback_interval: DEFAULT_FEEDBACK_INTERVAL) ⇒ void

Build and validate a logical replication stream configuration.

‘slot_name` and every publication name are intentionally limited to simple PostgreSQL identifier-like strings. This keeps command rendering small and predictable while avoiding quoting rules in this transport layer.

Boolean options are normalized with Ruby truthiness. ‘nil` and `false` become `false`; all other values become `true`.

Parameters:

  • database_url (#to_s) —

    PostgreSQL connection URL

  • slot_name (#to_s) —

    logical replication slot name

  • publication_names (Array<#to_s>, #to_s) —

    one or more publication names to pass to pgoutput

  • start_lsn (String, Integer, nil) (defaults to: nil) —

    starting LSN as a PostgreSQL LSN string, an integer WAL position, or ‘nil` for `0/0`

  • plugin (#to_s) (defaults to: DEFAULT_PLUGIN) —

    logical decoding plugin name

  • proto_version (#to_int, #to_s) (defaults to: DEFAULT_PROTO_VERSION) —

    pgoutput protocol version

  • binary (Object) (defaults to: false) —

    truthy to request binary column values

  • messages (Object) (defaults to: false) —

    truthy to request logical decoding messages

  • auto_create_slot (Object) (defaults to: false) —

    truthy to create the slot before starting replication

  • temporary_slot (Object) (defaults to: false) —

    truthy to create a temporary replication slot when ‘auto_create_slot` is enabled

  • feedback_interval (#to_f, #to_s) (defaults to: DEFAULT_FEEDBACK_INTERVAL) —

    seconds between periodic standby feedback messages

Raises:

  • (ConfigurationError) —

    if publication names are empty or numeric settings are invalid

  • (ArgumentError) —

    if ‘start_lsn`, `proto_version`, or `feedback_interval` cannot be coerced



125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
 
# File 'lib/pgoutput/client/configuration.rb', line 125

def initialize(database_url:,
               slot_name:,
               publication_names:,
               start_lsn: nil,
               plugin: DEFAULT_PLUGIN,
               proto_version: DEFAULT_PROTO_VERSION,
               binary: false,
               messages: false,
               auto_create_slot: false,
               temporary_slot: false,
               feedback_interval: DEFAULT_FEEDBACK_INTERVAL)
  @database_url = String(database_url).freeze
  @slot_name = validate_identifier(slot_name, "slot_name").freeze
  @publication_names = Array(publication_names).map do |name|
    validate_identifier(name, "publication_name").freeze
  end.freeze
  @start_lsn = normalize_lsn(start_lsn).freeze
  @plugin = String(plugin).freeze
  @proto_version = Integer(proto_version)
  @binary = boolean(binary, "binary")
  @messages = boolean(messages, "messages")
  @auto_create_slot = boolean(auto_create_slot, "auto_create_slot")
  @temporary_slot = boolean(temporary_slot, "temporary_slot")
  @feedback_interval = Float(feedback_interval)

  validate!
  freeze
end

Instance Attribute Details

#auto_create_slot ⇒ Boolean (readonly)

Whether the client should create the slot before streaming.

Returns:

  • (Boolean) 


82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/pgoutput/client/configuration.rb', line 82

attr_reader :database_url,
:slot_name,
:publication_names,
:start_lsn,
:plugin,
:proto_version,
:binary,
:messages,
:auto_create_slot,
:temporary_slot,
:feedback_interval

#binary ⇒ Boolean (readonly)

Whether to request binary column values from pgoutput.

Returns:

  • (Boolean) 


82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/pgoutput/client/configuration.rb', line 82

attr_reader :database_url,
:slot_name,
:publication_names,
:start_lsn,
:plugin,
:proto_version,
:binary,
:messages,
:auto_create_slot,
:temporary_slot,
:feedback_interval

#database_url ⇒ String (readonly)

PostgreSQL connection URL.

Returns:

  • (String) 


82
83
84
 
# File 'lib/pgoutput/client/configuration.rb', line 82

def database_url
  @database_url
end

#feedback_interval ⇒ Object (readonly) 



82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/pgoutput/client/configuration.rb', line 82

attr_reader :database_url,
:slot_name,
:publication_names,
:start_lsn,
:plugin,
:proto_version,
:binary,
:messages,
:auto_create_slot,
:temporary_slot,
:feedback_interval

#messages ⇒ Boolean (readonly)

Whether to request logical decoding messages from pgoutput.

Returns:

  • (Boolean) 


82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/pgoutput/client/configuration.rb', line 82

attr_reader :database_url,
:slot_name,
:publication_names,
:start_lsn,
:plugin,
:proto_version,
:binary,
:messages,
:auto_create_slot,
:temporary_slot,
:feedback_interval

#plugin ⇒ String (readonly)

Logical decoding output plugin name.

Returns:

  • (String) 


82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/pgoutput/client/configuration.rb', line 82

attr_reader :database_url,
:slot_name,
:publication_names,
:start_lsn,
:plugin,
:proto_version,
:binary,
:messages,
:auto_create_slot,
:temporary_slot,
:feedback_interval

#proto_version ⇒ Integer (readonly)

pgoutput protocol version.

Returns:

  • (Integer) 


82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/pgoutput/client/configuration.rb', line 82

attr_reader :database_url,
:slot_name,
:publication_names,
:start_lsn,
:plugin,
:proto_version,
:binary,
:messages,
:auto_create_slot,
:temporary_slot,
:feedback_interval

#publication_names ⇒ Array<String> (readonly)

Publication names requested from pgoutput.

Returns:

  • (Array<String>) 


82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/pgoutput/client/configuration.rb', line 82

attr_reader :database_url,
:slot_name,
:publication_names,
:start_lsn,
:plugin,
:proto_version,
:binary,
:messages,
:auto_create_slot,
:temporary_slot,
:feedback_interval

#slot_name ⇒ String (readonly)

Logical replication slot name.

Returns:

  • (String) 


82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/pgoutput/client/configuration.rb', line 82

attr_reader :database_url,
:slot_name,
:publication_names,
:start_lsn,
:plugin,
:proto_version,
:binary,
:messages,
:auto_create_slot,
:temporary_slot,
:feedback_interval

#start_lsn ⇒ String? (readonly)

Optional normalized starting LSN.

Returns:

  • (String, nil) 


82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/pgoutput/client/configuration.rb', line 82

attr_reader :database_url,
:slot_name,
:publication_names,
:start_lsn,
:plugin,
:proto_version,
:binary,
:messages,
:auto_create_slot,
:temporary_slot,
:feedback_interval

#temporary_slot ⇒ Boolean (readonly)

Whether a newly created slot should be temporary.

Returns:

  • (Boolean) 


82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/pgoutput/client/configuration.rb', line 82

attr_reader :database_url,
:slot_name,
:publication_names,
:start_lsn,
:plugin,
:proto_version,
:binary,
:messages,
:auto_create_slot,
:temporary_slot,
:feedback_interval

Instance Method Details

#start_lsn_string ⇒ String

Starting LSN to render in ‘START_REPLICATION`.

Returns:

  • (String) —

    normalized LSN string, defaulting to ‘“0/0”`



157
158
159
 
# File 'lib/pgoutput/client/configuration.rb', line 157

def start_lsn_string
  start_lsn || "0/0"
end