Top Fields Constructors Methods
org.postgresql.core.v3.replication

public Class V3PGReplicationStream

extends Object
implements PGReplicationStream
Class Inheritance
All Implemented Interfaces
org.postgresql.replication.PGReplicationStream, java.lang.AutoCloseable
Imports
org.postgresql.copy.CopyDual, org.postgresql.replication.LogSequenceNumber, .PGReplicationStream, .ReplicationType, org.postgresql.util.GT, .PSQLException, .PSQLState, org.checkerframework.checker.nullness.qual.Nullable, java.net.SocketTimeoutException, java.nio.ByteBuffer, java.sql.SQLException, java.util.Date, java.util.concurrent.TimeUnit, java.util.logging.Level, .Logger

Field Summary

Modifier and TypeField and Description
private boolean
closeFlag

private final CopyDual
copyDual

private volatile LogSequenceNumber
explicitlyFlushedLSN

private volatile LogSequenceNumber
lastAppliedLSN

private volatile LogSequenceNumber
lastFlushedLSN

private volatile LogSequenceNumber
lastReceiveLSN

Last receive LSN + payload size.

private LogSequenceNumber
lastServerLSN

private long
lastStatusUpdate

private static final Logger
LOGGER

private static final long
NANOS_PER_MILLISECOND

public static final long
POSTGRES_EPOCH_2000_01_01

private final ReplicationType
replicationType

private volatile LogSequenceNumber
startOfLastMessageLSN

private final long
updateInterval

Constructor Summary

AccessConstructor and Description
public
V3PGReplicationStream(CopyDual
bidirectional copy protocol
copyDual, LogSequenceNumber
the position in the WAL that we want to initiate replication from usually the currentLSN returned by calling pg_current_wal_lsn()for v10 above or pg_current_xlog_location() depending on the version of the server
startLSN, long
the number of millisecond between status packets sent back to the server. A value of zero disables the periodic status updates completely, although an update will still be sent when requested by the server, to avoid timeout disconnect.
updateIntervalMs, ReplicationType
LOGICAL or PHYSICAL
replicationType)

Method Summary

Modifier and TypeMethod and Description
private void
checkClose()

public void
close()

Implements org.postgresql.replication.PGReplicationStream.close.

Stop replication changes from server and free resources.

public void
forceUpdateStatus()

Implements org.postgresql.replication.PGReplicationStream.forceUpdateStatus.

Force send last received, flushed and applied LSN status to backend.

public LogSequenceNumber
getLastAppliedLSN()

Implements org.postgresql.replication.PGReplicationStream.getLastAppliedLSN.

Last applied lsn sent in update message to backed.

public LogSequenceNumber
getLastFlushedLSN()

Implements org.postgresql.replication.PGReplicationStream.getLastFlushedLSN.

Last flushed LSN sent in update message to backend.

public LogSequenceNumber
getLastReceiveLSN()

Implements org.postgresql.replication.PGReplicationStream.getLastReceiveLSN.

Parameter updates by execute PGReplicationStream#read() method.

public boolean
isClosed()

Implements org.postgresql.replication.PGReplicationStream.isClosed.

private boolean
isTimeUpdate()

private byte[]
prepareUpdateStatus(LogSequenceNumber received, LogSequenceNumber flushed, LogSequenceNumber applied, boolean replyRequired)

private boolean
processKeepAliveMessage(ByteBuffer buffer)

private ByteBuffer
processXLogData(ByteBuffer buffer)

public @Nullable ByteBuffer
read()

Implements org.postgresql.replication.PGReplicationStream.read.

Read next wal record from backend.

private @Nullable ByteBuffer
readInternal(boolean block)

public @Nullable ByteBuffer
readPending()

Implements org.postgresql.replication.PGReplicationStream.readPending.

Read next WAL record from backend.

private @Nullable ByteBuffer
receiveNextData(boolean block)

public void
setAppliedLSN(LogSequenceNumber
NOT NULL location of the last WAL applied in the standby.
applied)

Implements org.postgresql.replication.PGReplicationStream.setAppliedLSN.

Inform backend which LSN has been applied on standby.

public void
setFlushedLSN(LogSequenceNumber
NOT NULL location of the last WAL flushed to disk in the standby.
flushed)

Implements org.postgresql.replication.PGReplicationStream.setFlushedLSN.

Set flushed LSN.

private void
timeUpdateStatus()

private void
updateStatusInternal(LogSequenceNumber received, LogSequenceNumber flushed, LogSequenceNumber applied, boolean replyRequired)

Inherited from java.lang.Object:
cloneequalsfinalizegetClasshashCodenotifynotifyAlltoStringwaitwaitwait

Field Detail

closeFlagback to summary
private boolean closeFlag
copyDualback to summary
private final CopyDual copyDual
explicitlyFlushedLSNback to summary
private volatile LogSequenceNumber explicitlyFlushedLSN
lastAppliedLSNback to summary
private volatile LogSequenceNumber lastAppliedLSN
lastFlushedLSNback to summary
private volatile LogSequenceNumber lastFlushedLSN
lastReceiveLSNback to summary
private volatile LogSequenceNumber lastReceiveLSN

Last receive LSN + payload size.

lastServerLSNback to summary
private LogSequenceNumber lastServerLSN
lastStatusUpdateback to summary
private long lastStatusUpdate
LOGGERback to summary
private static final Logger LOGGER
NANOS_PER_MILLISECONDback to summary
private static final long NANOS_PER_MILLISECOND
POSTGRES_EPOCH_2000_01_01back to summary
public static final long POSTGRES_EPOCH_2000_01_01
replicationTypeback to summary
private final ReplicationType replicationType
startOfLastMessageLSNback to summary
private volatile LogSequenceNumber startOfLastMessageLSN
updateIntervalback to summary
private final long updateInterval

Constructor Detail

V3PGReplicationStreamback to summary
public V3PGReplicationStream(CopyDual copyDual, LogSequenceNumber startLSN, long updateIntervalMs, ReplicationType replicationType)
Parameters
copyDual:CopyDual

bidirectional copy protocol

startLSN:LogSequenceNumber

the position in the WAL that we want to initiate replication from usually the currentLSN returned by calling pg_current_wal_lsn()for v10 above or pg_current_xlog_location() depending on the version of the server

updateIntervalMs:long

the number of millisecond between status packets sent back to the server. A value of zero disables the periodic status updates completely, although an update will still be sent when requested by the server, to avoid timeout disconnect.

replicationType:ReplicationType

LOGICAL or PHYSICAL

Method Detail

checkCloseback to summary
private void checkClose() throws PSQLException
closeback to summary
public void close() throws SQLException

Implements org.postgresql.replication.PGReplicationStream.close.

Doc from org.postgresql.replication.PGReplicationStream.close.

Stop replication changes from server and free resources. After that connection can be reuse to another queries. Also after close current stream they cannot be used anymore.

Note

This method can spend much time for logical replication stream on postgresql version 9.6 and lower, because postgresql have bug - during decode big transaction to logical form and during wait new changes postgresql ignore messages from client. As workaround you can close replication connection instead of close replication stream. For more information about it problem see mailing list thread Stopping logical replication protocol

Annotations
@Override
Exceptions
SQLException:
when some internal exception occurs during end streaming
forceUpdateStatusback to summary
public void forceUpdateStatus() throws SQLException

Implements org.postgresql.replication.PGReplicationStream.forceUpdateStatus.

Doc from org.postgresql.replication.PGReplicationStream.forceUpdateStatus.

Force send last received, flushed and applied LSN status to backend. You cannot send LSN status explicitly because PGReplicationStream sends the status to backend periodically by configured interval via LogicalReplicationOptions#getStatusInterval

Annotations
@Override
Exceptions
SQLException:
when some internal exception occurs during read from stream
getLastAppliedLSNback to summary
public LogSequenceNumber getLastAppliedLSN()

Implements org.postgresql.replication.PGReplicationStream.getLastAppliedLSN.

Doc from org.postgresql.replication.PGReplicationStream.getLastAppliedLSN.

Last applied lsn sent in update message to backed. Parameter updates only via PGReplicationStream#setAppliedLSN(LogSequenceNumber)

It is safe to call this method in a thread different than the main thread.

Returns:LogSequenceNumber

not null location of the last WAL applied in the standby.

Annotations
@Override
getLastFlushedLSNback to summary
public LogSequenceNumber getLastFlushedLSN()

Implements org.postgresql.replication.PGReplicationStream.getLastFlushedLSN.

Doc from org.postgresql.replication.PGReplicationStream.getLastFlushedLSN.

Last flushed LSN sent in update message to backend. Parameter updates only via PGReplicationStream#setFlushedLSN(LogSequenceNumber)

It is safe to call this method in a thread different than the main thread.

Returns:LogSequenceNumber

NOT NULL location of the last WAL flushed to disk in the standby.

Annotations
@Override
getLastReceiveLSNback to summary
public LogSequenceNumber getLastReceiveLSN()

Implements org.postgresql.replication.PGReplicationStream.getLastReceiveLSN.

Doc from org.postgresql.replication.PGReplicationStream.getLastReceiveLSN.

Parameter updates by execute PGReplicationStream#read() method.

It is safe to call this method in a thread different than the main thread. However, usually this method is called in the main thread after a successful PGReplicationStream#read() or PGReplicationStream#readPending(), to get the LSN corresponding to the received record.

Returns:LogSequenceNumber

NOT NULL LSN position that was receive last time via PGReplicationStream#read() method

Annotations
@Override
isClosedback to summary
public boolean isClosed()

Implements org.postgresql.replication.PGReplicationStream.isClosed.

Returns:boolean

Doc from org.postgresql.replication.PGReplicationStream.isClosed.

true if replication stream was already close, otherwise return false

Annotations
@Override
isTimeUpdateback to summary
private boolean isTimeUpdate()
prepareUpdateStatusback to summary
private byte[] prepareUpdateStatus(LogSequenceNumber received, LogSequenceNumber flushed, LogSequenceNumber applied, boolean replyRequired)
processKeepAliveMessageback to summary
private boolean processKeepAliveMessage(ByteBuffer buffer)
processXLogDataback to summary
private ByteBuffer processXLogData(ByteBuffer buffer)
readback to summary
public @Nullable ByteBuffer read() throws SQLException

Implements org.postgresql.replication.PGReplicationStream.read.

Doc from org.postgresql.replication.PGReplicationStream.read.

Read next wal record from backend. It method can be block until new message will not get from server.

A single WAL record is never split across two XLogData messages. When a WAL record crosses a WAL page boundary, and is therefore already split using continuation records, it can be split at the page boundary. In other words, the first main WAL record and its continuation records can be sent in different XLogData messages.

Returns:@Nullable ByteBuffer

not null byte array received by replication protocol, return ByteBuffer wrap around received byte array with use offset, so, use ByteBuffer#array() carefully

Annotations
@Override
Exceptions
SQLException:
when some internal exception occurs during read from stream
readInternalback to summary
private @Nullable ByteBuffer readInternal(boolean block) throws SQLException
readPendingback to summary
public @Nullable ByteBuffer readPending() throws SQLException

Implements org.postgresql.replication.PGReplicationStream.readPending.

Doc from org.postgresql.replication.PGReplicationStream.readPending.

Read next WAL record from backend. This method does not block and in contrast to PGReplicationStream#read(). If message from backend absent return null. It allow periodically check message in stream and if they absent sleep some time, but it time should be less than CommonOptions#getStatusInterval() to avoid disconnect from the server.

A single WAL record is never split across two XLogData messages. When a WAL record crosses a WAL page boundary, and is therefore already split using continuation records, it can be split at the page boundary. In other words, the first main WAL record and its continuation records can be sent in different XLogData messages.

Returns:@Nullable ByteBuffer

byte array received by replication protocol or NULL if pending message from server absent. Returns ByteBuffer wrap around received byte array with use offset, so, use ByteBuffer#array() carefully.

Annotations
@Override
Exceptions
SQLException:
when some internal exception occurs during read from stream
receiveNextDataback to summary
private @Nullable ByteBuffer receiveNextData(boolean block) throws SQLException
setAppliedLSNback to summary
public void setAppliedLSN(LogSequenceNumber applied)

Implements org.postgresql.replication.PGReplicationStream.setAppliedLSN.

Doc from org.postgresql.replication.PGReplicationStream.setAppliedLSN.

Inform backend which LSN has been applied on standby. Feedback will send to backend on next update status iteration.

It is safe to call this method in a thread different than the main thread. The updated value will be sent to the backend in the next status update run.

Parameters
applied:LogSequenceNumber

NOT NULL location of the last WAL applied in the standby.

Annotations
@Override
setFlushedLSNback to summary
public void setFlushedLSN(LogSequenceNumber flushed)

Implements org.postgresql.replication.PGReplicationStream.setFlushedLSN.

Doc from org.postgresql.replication.PGReplicationStream.setFlushedLSN.

Set flushed LSN. This parameter will be sent to backend on next update status iteration. Flushed LSN position help backend define which WAL can be recycled.

It is safe to call this method in a thread different than the main thread. The updated value will be sent to the backend in the next status update run.

Parameters
flushed:LogSequenceNumber

NOT NULL location of the last WAL flushed to disk in the standby.

Annotations
@Override
timeUpdateStatusback to summary
private void timeUpdateStatus() throws SQLException
updateStatusInternalback to summary
private void updateStatusInternal(LogSequenceNumber received, LogSequenceNumber flushed, LogSequenceNumber applied, boolean replyRequired) throws SQLException