Class: Pgoutput::Client::Stream Private

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

Overview

This class is part of a private API. You should avoid using this class if possible, as it may be removed or be changed in the future.

Logical replication stream loop.

Stream consumes PostgreSQL CopyData payloads from a Connection. It understands the two replication envelope message types used by PostgreSQL's streaming replication protocol:

  • w — XLogData, containing logical decoding plugin payload bytes.
  • k — primary keepalive, optionally requesting immediate feedback.

The stream yields only XLogData plugin payloads. Keepalive messages are handled internally by updating the latest known WAL position and sending standby feedback when requested.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(connection:, configuration:, acked_lsn: nil) ⇒ Stream

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Build a stream loop.

Parameters:

  • connection (Connection) —

    replication connection

  • configuration (Configuration) —

    stream configuration

  • connection: (Object)
  • configuration: (Object)
  • acked_lsn: (Object) (defaults to: nil) 


48
49
50
51
52
53
54
55
56
57
 
# File 'lib/pgoutput/client/stream.rb', line 48

def initialize(connection:, configuration:, acked_lsn: nil)
  @connection = connection
  @configuration = configuration
  @latest_lsn = LSN.parse(configuration.start_lsn_string)
  @acked_lsn = acked_lsn ? normalize_lsn_value(acked_lsn) : @latest_lsn
  @last_feedback_at = monotonic_time
  @last_keepalive_at = nil
  @running = false
  @stop_requested = false
end

Instance Attribute Details

#acked_lsn ⇒ Integer (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Latest downstream-acknowledged WAL position used as flushed/applied LSN in standby feedback.

Returns:

  • (Integer) 


34
35
36
 
# File 'lib/pgoutput/client/stream.rb', line 34

def acked_lsn
  @acked_lsn
end

#last_keepalive_at ⇒ Time? (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Last time a primary keepalive was observed.

Returns:

  • (Time, nil) 


39
40
41
 
# File 'lib/pgoutput/client/stream.rb', line 39

def last_keepalive_at
  @last_keepalive_at
end

#latest_lsn ⇒ Integer (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Latest confirmed WAL position observed by this stream.

Returns:

  • (Integer) 


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

def latest_lsn
  @latest_lsn
end

Instance Method Details

#ack(lsn) ⇒ Integer

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Mark a WAL position as durably handled by downstream code.

The stream never decides sink durability on its own. Downstream code may call this after checkpointing, enqueueing, or otherwise making progress durable. Feedback sent after this call reports the acknowledged LSN as both flushed and applied.

Parameters:

  • lsn (String, Integer) —

    WAL position acknowledged by downstream code

Returns:

  • (Integer) —

    normalized acknowledged WAL position



119
120
121
122
 
# File 'lib/pgoutput/client/stream.rb', line 119

def ack(lsn)
  parsed = normalize_lsn_value(lsn)
  @acked_lsn = [@acked_lsn, parsed].max
end

#monotonic_time ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns:

  • (Object) 


159
160
161
 
# File 'lib/pgoutput/client/stream.rb', line 159

def monotonic_time
  Process.clock_gettime(Process::CLOCK_MONOTONIC)
end

#normalize_lsn_value(value) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Parameters:

  • value (Object)

Returns:

  • (Object) 


155
156
157
 
# File 'lib/pgoutput/client/stream.rb', line 155

def normalize_lsn_value(value)
  value.is_a?(String) ? LSN.parse(value) : LSN.parse(LSN.format(value))
end

#process_copy_data(copy_data) {|arg0, arg1| ... } ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Parameters:

  • copy_data (Object)

Yields:

Yield Parameters:

  • arg0 (Object)
  • arg1 (Object)

Yield Returns:

  • (Object)

Returns:

  • (Object) 


126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
 
# File 'lib/pgoutput/client/stream.rb', line 126

def process_copy_data(copy_data)
  case copy_data.getbyte(0)
  when "w".ord
    xlog = XLogData.parse(copy_data)
    @latest_lsn = xlog.wal_end
    yield xlog.payload, xlog
  when "k".ord
    keepalive = Keepalive.parse(copy_data)
    @latest_lsn = [@latest_lsn, keepalive.wal_end].max
    @last_keepalive_at = Time.now.utc
    send_feedback(reply_requested: true) if keepalive.reply_requested
  else
    raise ProtocolError, "unknown CopyData replication message: #{copy_data.getbyte(0).inspect}"
  end
end

#running? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Whether the stream loop is active.

Returns:

  • (Boolean) 


106
107
108
 
# File 'lib/pgoutput/client/stream.rb', line 106

def running?
  @running
end

#send_feedback(reply_requested:) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Parameters:

  • reply_requested: (Object)

Returns:

  • (Object) 


148
149
150
151
152
153
 
# File 'lib/pgoutput/client/stream.rb', line 148

def send_feedback(reply_requested:)
  feedback = Feedback.now(received_lsn: @latest_lsn, flushed_lsn: @acked_lsn, applied_lsn: @acked_lsn,
                          reply_requested:)
  @connection.put_copy_data(feedback.to_copy_data)
  @last_feedback_at = monotonic_time
end

#send_periodic_feedback ⇒ nil, untyped

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns:

  • (nil, untyped) 


142
143
144
145
146
 
# File 'lib/pgoutput/client/stream.rb', line 142

def send_periodic_feedback
  return if monotonic_time - @last_feedback_at < @configuration.feedback_interval

  send_feedback(reply_requested: false)
end

#start {|payload, metadata| ... } ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Start the stream loop.

The method blocks while the stream is running. For every XLogData envelope, it yields the raw pgoutput payload and the parsed envelope metadata. When no CopyData payload is currently available, the loop pauses briefly before checking again.

Yields:

  • (payload, metadata) —

    called for each XLogData payload

Yield Parameters:

  • payload (String) —

    frozen raw pgoutput payload bytes

  • metadata (XLogData) —

    parsed WAL envelope metadata

Raises:

  • (ArgumentError) —

    if no block is provided

  • (ProtocolError) —

    if an unknown or malformed replication message is received

  • (ConnectionError) —

    if standby feedback cannot be sent



74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/pgoutput/client/stream.rb', line 74

def start(&)
  raise ArgumentError, "block required" unless block_given?

  @running = true
  while @running
    copy_data = @connection.get_copy_data
    if copy_data.nil?
      send_periodic_feedback
      sleep 0.01
      next
    end

    process_copy_data(copy_data, &)
    send_periodic_feedback
  end
ensure
  send_feedback(reply_requested: false) if @stop_requested
  @running = false
end

#stop ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Stop the stream loop after the current iteration.



97
98
99
100
101
 
# File 'lib/pgoutput/client/stream.rb', line 97

def stop
  @stop_requested = true
  @running = false
  nil
end