Class: Pgoutput::Decoder

Inherits:
Object
  • Object
show all
Defined in:
lib/pgoutput/decoder.rb,
lib/pgoutput/decoder/errors.rb,
lib/pgoutput/decoder/events.rb,
lib/pgoutput/decoder/version.rb,
lib/pgoutput/decoder/row_builder.rb,
lib/pgoutput/decoder/type_registry.rb,
lib/pgoutput/decoder/value_decoder.rb,
lib/pgoutput/decoder/relation_cache.rb,
sig/pgoutput_decoder.rbs

Overview

Stateful high-level decoder for pgoutput-parser protocol messages.

Decoder accepts immutable protocol messages from pgoutput-parser and returns immutable, Ractor-shareable row-change events. The decoder maintains relation and active transaction context, so one instance should be used per logical replication stream.

Defined Under Namespace

Modules: Events Classes: Error, RelationCache, RowBuilder, TransactionStateError, TypeRegistry, UnknownRelationError, UnsupportedMessageError, ValueDecodeError, ValueDecoder

Constant Summary collapse

VERSION =

Gem version.

Returns:

  • (String) 
"0.2.0"

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(type_registry: TypeRegistry.default) ⇒ Decoder

Returns a new instance of Decoder.

Parameters:

  • type_registry (TypeRegistry) (defaults to: TypeRegistry.default) —

    immutable OID decoder registry.

  • type_registry: (TypeRegistry) (defaults to: TypeRegistry.default) 


32
33
34
35
36
37
38
39
 
# File 'lib/pgoutput/decoder.rb', line 32

def initialize(type_registry: TypeRegistry.default)
  @type_registry = type_registry
  @relations = RelationCache.new
  @row_builder = RowBuilder.new(type_registry: type_registry)
  @current_transaction_id = nil
  @current_final_lsn = nil
  @current_commit_timestamp = nil
end

Instance Attribute Details

#type_registry ⇒ TypeRegistry (readonly)

Returns:



28
29
30
 
# File 'lib/pgoutput/decoder.rb', line 28

def type_registry
  @type_registry
end

Instance Method Details

#clear_transaction! ⇒ nil

Returns:

  • (nil) 


161
162
163
164
165
 
# File 'lib/pgoutput/decoder.rb', line 161

def clear_transaction!
  @current_transaction_id = nil
  @current_final_lsn = nil
  @current_commit_timestamp = nil
end

#decode(message) ⇒ Events::Begin, ...

Decode one pgoutput-parser protocol message.

Parameters:

  • message (Object) —

    protocol message from pgoutput-parser.

Returns:

Raises:



47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/pgoutput/decoder.rb', line 47

def decode(message)
  case message
  when parser_messages::Begin
    decode_begin(message)
  when parser_messages::Relation
    @relations.store(message)
    nil
  when parser_messages::Insert
    decode_insert(message)
  when parser_messages::Update
    decode_update(message)
  when parser_messages::Delete
    decode_delete(message)
  when parser_messages::Commit
    decode_commit(message)
  else
    raise UnsupportedMessageError, "unsupported message: #{message.class}"
  end
end

#decode_begin(message) ⇒ Events::Begin

Parameters:

  • message (parser_message)

Returns:



69
70
71
72
73
74
75
76
77
78
79
80
81
 
# File 'lib/pgoutput/decoder.rb', line 69

def decode_begin(message)
  @current_transaction_id = message.xid
  @current_final_lsn = message.final_lsn
  @current_commit_timestamp = message.commit_timestamp

  share(
    Events::Begin.new(
      message.xid,
      message.final_lsn,
      message.commit_timestamp
    )
  )
end

#decode_commit(message) ⇒ Events::Commit

Parameters:

  • message (parser_message)

Returns:



83
84
85
86
87
88
89
90
91
92
93
94
95
96
 
# File 'lib/pgoutput/decoder.rb', line 83

def decode_commit(message)
  transaction_id = @current_transaction_id

  event = Events::Commit.new(
    transaction_id,
    message.flags,
    message.commit_lsn,
    message.transaction_end_lsn,
    message.commit_timestamp
  )

  clear_transaction!
  share(event)
end

#decode_delete(message) ⇒ Events::Delete

Parameters:

  • message (parser_message)

Returns:



130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
 
# File 'lib/pgoutput/decoder.rb', line 130

def decode_delete(message)
  relation = relation_for(message.relation_id)
  transaction_id = require_transaction_id

  share(
    Events::Delete.new(
      transaction_id,
      message.relation_id,
      relation.schema,
      relation.table,
      optional_row(relation, message.old_key_tuple, key: true),
      optional_row(relation, message.old_tuple)
    )
  )
end

#decode_insert(message) ⇒ Events::Insert

Parameters:

  • message (parser_message)

Returns:



98
99
100
101
102
103
104
105
106
107
108
109
110
111
 
# File 'lib/pgoutput/decoder.rb', line 98

def decode_insert(message)
  relation = relation_for(message.relation_id)
  transaction_id = require_transaction_id

  share(
    Events::Insert.new(
      transaction_id,
      message.relation_id,
      relation.schema,
      relation.table,
      @row_builder.build(relation, message.tuple)
    )
  )
end

#decode_update(message) ⇒ Events::Update

Parameters:

  • message (parser_message)

Returns:



113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
 
# File 'lib/pgoutput/decoder.rb', line 113

def decode_update(message)
  relation = relation_for(message.relation_id)
  transaction_id = require_transaction_id

  share(
    Events::Update.new(
      transaction_id,
      message.relation_id,
      relation.schema,
      relation.table,
      optional_row(relation, message.old_key_tuple, key: true),
      optional_row(relation, message.old_tuple),
      @row_builder.build(relation, message.new_tuple)
    )
  )
end

#optional_row(relation, tuple, key: false) ⇒ Hash[String, untyped]?

Parameters:

  • relation (relation)
  • tuple (tuple, nil)
  • key: (Boolean) (defaults to: false)

Returns:

  • (Hash[String, untyped], nil) 


146
147
148
149
150
151
 
# File 'lib/pgoutput/decoder.rb', line 146

def optional_row(relation, tuple, key: false)
  return nil if tuple.nil?
  return @row_builder.build_key(relation, tuple) if key

  @row_builder.build(relation, tuple)
end

#parser_messages ⇒ Object

Returns:

  • (Object) 


167
168
169
 
# File 'lib/pgoutput/decoder.rb', line 167

def parser_messages
  Pgoutput::Messages
end

#relation_for(relation_id) ⇒ relation

Parameters:

  • relation_id (Integer)

Returns:

  • (relation) 


153
154
155
 
# File 'lib/pgoutput/decoder.rb', line 153

def relation_for(relation_id)
  @relations.fetch(relation_id)
end

#require_transaction_id ⇒ Integer

Returns:

  • (Integer) 


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

def require_transaction_id
  @current_transaction_id || raise(TransactionStateError, "DML message received outside an active transaction")
end

#share(object) ⇒ void

This method returns an undefined value.



171
172
173
 
# File 'lib/pgoutput/decoder.rb', line 171

def share(object)
  Ractor.make_shareable(object)
end