Class: Pgoutput::SourceAdapter::Cdc

Inherits:
CDC::Core::SourceAdapter
  • Object
show all
Defined in:
lib/pgoutput/source_adapter/cdc.rb

Overview

Normalizes Pgoutput::Decoder::Events into CDC::Core primitives.

This adapter is intentionally located under the Pgoutput namespace because it adapts pgoutput decoded events. The target is CDC::Core, so this class depends on cdc-core while the lower-level pgoutput-client, pgoutput-parser, and pgoutput-decoder gems remain standalone.

rubocop:disable Metrics/ClassLength

Examples:

Normalize a decoded insert event

adapter = Pgoutput::SourceAdapter::Cdc.new
change_event = adapter.normalize(decoded_insert)

Normalize a decoded transaction event batch

envelope = adapter.normalize_many([begin_event, insert_event, commit_event]).first

Normalize an unbounded decoded event stream

adapter.each_normalized(decoded_events) { |work| process(work) }

Defined Under Namespace

Classes: StreamEvent

Constant Summary collapse

SOURCE_NAME =

Canonical source name included in normalized event metadata.

Returns:

  • (String) 
'pgoutput'

Instance Method Summary collapse

Constructor Details

#initialize(primary_key_resolver: nil, metadata_builder: nil) ⇒ void

Parameters:

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

    optional callable used to infer primary keys from decoded row values when pgoutput does not provide an old-key tuple. The callable receives the decoded event and value hash.

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

    optional callable that can return extra metadata for each decoded event. Returned keys are stringified.



73
74
75
76
77
 
# File 'lib/pgoutput/source_adapter/cdc.rb', line 73

def initialize(primary_key_resolver: nil, metadata_builder: nil)
  @primary_key_resolver = primary_key_resolver || method(:default_primary_key)
  @metadata_builder = metadata_builder
  super()
end

Instance Method Details

#each_normalized(events) {|work| ... } ⇒ Enumerator?

Incrementally normalize a decoded pgoutput event stream.

Transaction buffering belongs to this source adapter: row changes are retained after Begin and emitted as one CDC::Core::TransactionEnvelope at Commit. Changes outside a transaction are yielded immediately.

Parameters:

  • events (Enumerable<Object, StreamEvent>) —

    decoded stream

Yield Parameters:

  • work (CDC::Core::ChangeEvent, CDC::Core::TransactionEnvelope)

Returns:

  • (Enumerator, nil) —

    enumerator without a block, otherwise nil



155
156
157
158
159
160
 
# File 'lib/pgoutput/source_adapter/cdc.rb', line 155

def each_normalized(events, &block)
  return enum_for(:each_normalized, events) unless block

  normalize_stream(events, flush_incomplete: false, &block)
  nil
end

#normalize(event, source_position: nil, sequence_number: nil) ⇒ CDC::Core::ChangeEvent?

Normalize one decoded pgoutput event.

Transaction boundary events return nil because they do not represent a row-level change by themselves. Use #normalize_many when transaction envelopes are desired.

rubocop:disable Metrics/MethodLength

Parameters:

  • event (Object) —

    a Pgoutput::Decoder::Events object.

  • sequence_number (Integer, nil) (defaults to: nil) —

    stable zero-based ordinal within the event's transaction.

Returns:

  • (CDC::Core::ChangeEvent, nil) —

    normalized row change event, or nil for transaction boundary events.

Raises:



93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
 
# File 'lib/pgoutput/source_adapter/cdc.rb', line 93

def normalize(event, source_position: nil, sequence_number: nil)
  case event_name(event)
  when 'Insert'
    change_event(
      event,
      operation: :insert,
      old_values: nil,
      new_values: event.values,
      primary_key: primary_key_for(event, event.values),
      source_position: source_position,
      sequence_number: sequence_number
    )
  when 'Update'
    old_values = event.old_values || event.old_key

    change_event(
      event,
      operation: :update,
      old_values: old_values,
      new_values: event.new_values,
      primary_key: primary_key_for(event, event.new_values),
      source_position: source_position,
      sequence_number: sequence_number
    )
  when 'Delete'
    old_values = event.old_values || event.old_key

    change_event(
      event,
      operation: :delete,
      old_values: old_values,
      new_values: nil,
      primary_key: primary_key_for(event, old_values),
      source_position: source_position,
      sequence_number: sequence_number
    )
  when 'Begin', 'Commit'
    nil
  else
    raise Error, "unsupported pgoutput decoded event: #{event.class}"
  end
end

#normalize_many(events) ⇒ Array<CDC::Core::ChangeEvent, CDC::Core::TransactionEnvelope>

Normalize a sequence of decoded pgoutput events.

If the sequence contains transaction boundaries, row changes between a Begin and Commit are grouped into CDC::Core::TransactionEnvelope. If no transaction boundaries are present, row changes are returned individually.

Parameters:

  • events (Enumerable<Object>) —

    decoded pgoutput events.

Returns:

  • (Array<CDC::Core::ChangeEvent, CDC::Core::TransactionEnvelope>) —

    normalized row changes and transaction envelopes in input order.

Raises:



173
174
175
176
177
 
# File 'lib/pgoutput/source_adapter/cdc.rb', line 173

def normalize_many(events)
  results = [] # : Array[CDC::Core::ChangeEvent | CDC::Core::TransactionEnvelope]
  normalize_stream(events, flush_incomplete: true) { |work| results << work }
  share(results.freeze)
end

#stream_event(event, source_position: nil) ⇒ StreamEvent

Pair a decoded event with the source position supplied by its transport.

Parameters:

  • event (Object) —

    decoded pgoutput event

  • source_position (Object, nil) (defaults to: nil) —

    WAL position associated with the event

Returns:



142
143
144
 
# File 'lib/pgoutput/source_adapter/cdc.rb', line 142

def stream_event(event, source_position: nil)
  StreamEvent.new(event, lsn_string(source_position))
end