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



150
151
152
153
154
155
 
# File 'lib/pgoutput/source_adapter/cdc.rb', line 150

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) ⇒ 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.

Returns:

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

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

Raises:



91
92
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
 
# File 'lib/pgoutput/source_adapter/cdc.rb', line 91

def normalize(event, source_position: 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
    )
  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
    )
  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
    )
  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:



168
169
170
171
172
 
# File 'lib/pgoutput/source_adapter/cdc.rb', line 168

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:



137
138
139
 
# File 'lib/pgoutput/source_adapter/cdc.rb', line 137

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