#include "postgres_fe.h"
#include <dirent.h>
#include <limits.h>
#include <sys/select.h>
#include <sys/stat.h>
#include <unistd.h>
#include "common/file_perm.h"
#include "common/logging.h"
#include "fe_utils/option_utils.h"
#include "getopt_long.h"
#include "libpq-fe.h"
#include "libpq/pqsignal.h"
#include "libpq/protocol.h"
#include "pqexpbuffer.h"
#include "streamutil.h"

Include dependency graph for pg_recvlogical.c:

Macros
#define	RECONNECT_SLEEP_TIME 5

Enumerations
enum	StreamStopReason { STREAM_STOP_NONE , STREAM_STOP_END_OF_WAL , STREAM_STOP_KEEPALIVE , STREAM_STOP_SIGNAL }

Functions
static void	usage (void)

static void	StreamLogicalLog (void)

static bool	flushAndSendFeedback (PGconn conn, TimestampTz now)

static void	prepareToTerminate (PGconn *conn, XLogRecPtr endpos, StreamStopReason reason, XLogRecPtr lsn)

static bool	sendFeedback (PGconn *conn, TimestampTz now, bool force, bool replyRequested)

static void	disconnect_atexit (void)

static void	OutputFsync (TimestampTz now)

static void	sigexit_handler (SIGNAL_ARGS)

static void	sighup_handler (SIGNAL_ARGS)

int	main (int argc, char **argv)

Variables
static char *	outfile = NULL

static int	verbose = 0

static bool	two_phase = false

static bool	failover = false

static int	noloop = 0

static int	standby_message_timeout = 10 * 1000

static int	fsync_interval = 10 * 1000

static XLogRecPtr	startpos = InvalidXLogRecPtr

static XLogRecPtr	endpos = InvalidXLogRecPtr

static bool	do_create_slot = false

static bool	slot_exists_ok = false

static bool	do_start_slot = false

static bool	do_drop_slot = false

static char *	replication_slot = NULL

static char **	options

static size_t	noptions = 0

static const char *	plugin = "test_decoding"

static int	outfd = -1

static volatile sig_atomic_t	time_to_abort = false

static volatile sig_atomic_t	stop_reason = STREAM_STOP_NONE

static volatile sig_atomic_t	output_reopen = false

static bool	output_isfile

static TimestampTz	output_last_fsync = -1

static bool	output_needs_fsync = false

static XLogRecPtr	output_written_lsn = InvalidXLogRecPtr

static XLogRecPtr	output_fsync_lsn = InvalidXLogRecPtr

Macro Definition Documentation

◆ RECONNECT_SLEEP_TIME

#define RECONNECT_SLEEP_TIME 5

Definition at line 32 of file pg_recvlogical.c.

Enumeration Type Documentation

◆ StreamStopReason

enum StreamStopReason

Enumerator
STREAM_STOP_NONE
STREAM_STOP_END_OF_WAL
STREAM_STOP_KEEPALIVE
STREAM_STOP_SIGNAL

Definition at line 34 of file pg_recvlogical.c.

{
    STREAM_STOP_NONE,
    STREAM_STOP_END_OF_WAL,
    STREAM_STOP_KEEPALIVE,
    STREAM_STOP_SIGNAL
} StreamStopReason;

Function Documentation

◆ disconnect_atexit()

static void disconnect_atexit ( void )

static

Definition at line 181 of file pg_recvlogical.c.

{
    if (conn != NULL)
        PQfinish(conn);
}

References conn, fb(), and PQfinish().

Referenced by main().

◆ flushAndSendFeedback()

static bool flushAndSendFeedback	(	PGconn *	conn,
		TimestampTz *	now
	)

static

Definition at line 1058 of file pg_recvlogical.c.

{
    /* flush data to disk, so that we send a recent flush pointer */
    OutputFsync(*now);
    *now = feGetCurrentTimestamp();
    if (!sendFeedback(conn, *now, true, false))
        return false;
 
    return true;
}

References conn, feGetCurrentTimestamp(), now(), OutputFsync(), and sendFeedback().

Referenced by prepareToTerminate(), and StreamLogicalLog().

◆ main()

int main	(	int	argc,
		char **	argv
	)

Definition at line 696 of file pg_recvlogical.c.

{
    static struct option long_options[] = {
/* general options */
        {"file", required_argument, NULL, 'f'},
        {"fsync-interval", required_argument, NULL, 'F'},
        {"no-loop", no_argument, NULL, 'n'},
        {"enable-failover", no_argument, NULL, 5},
        {"enable-two-phase", no_argument, NULL, 't'},
        {"two-phase", no_argument, NULL, 't'},  /* deprecated */
        {"verbose", no_argument, NULL, 'v'},
        {"version", no_argument, NULL, 'V'},
        {"help", no_argument, NULL, '?'},
/* connection options */
        {"dbname", required_argument, NULL, 'd'},
        {"host", required_argument, NULL, 'h'},
        {"port", required_argument, NULL, 'p'},
        {"username", required_argument, NULL, 'U'},
        {"no-password", no_argument, NULL, 'w'},
        {"password", no_argument, NULL, 'W'},
/* replication options */
        {"startpos", required_argument, NULL, 'I'},
        {"endpos", required_argument, NULL, 'E'},
        {"option", required_argument, NULL, 'o'},
        {"plugin", required_argument, NULL, 'P'},
        {"status-interval", required_argument, NULL, 's'},
        {"slot", required_argument, NULL, 'S'},
/* action */
        {"create-slot", no_argument, NULL, 1},
        {"start", no_argument, NULL, 2},
        {"drop-slot", no_argument, NULL, 3},
        {"if-not-exists", no_argument, NULL, 4},
        {NULL, 0, NULL, 0}
    };
    int         c;
    int         option_index;
    uint32      hi,
                lo;
    char       *db_name;
 
    pg_logging_init(argv[0]);
    progname = get_progname(argv[0]);
    set_pglocale_pgservice(argv[0], PG_TEXTDOMAIN("pg_basebackup"));
 
    if (argc > 1)
    {
        if (strcmp(argv[1], "--help") == 0 || strcmp(argv[1], "-?") == 0)
        {
            usage();
            exit(0);
        }
        else if (strcmp(argv[1], "-V") == 0 ||
                 strcmp(argv[1], "--version") == 0)
        {
            puts("pg_recvlogical (PostgreSQL) " PG_VERSION);
            exit(0);
        }
    }
 
    while ((c = getopt_long(argc, argv, "E:f:F:ntvd:h:p:U:wWI:o:P:s:S:",
                            long_options, &option_index)) != -1)
    {
        switch (c)
        {
/* general options */
            case 'f':
                outfile = pg_strdup(optarg);
                break;
            case 'F':
                if (!option_parse_int(optarg, "-F/--fsync-interval", 0,
                                      INT_MAX / 1000,
                                      &fsync_interval))
                    exit(1);
                fsync_interval *= 1000;
                break;
            case 'n':
                noloop = 1;
                break;
            case 't':
                two_phase = true;
                break;
            case 'v':
                verbose++;
                break;
            case 5:
                failover = true;
                break;
/* connection options */
            case 'd':
                dbname = pg_strdup(optarg);
                break;
            case 'h':
                dbhost = pg_strdup(optarg);
                break;
            case 'p':
                dbport = pg_strdup(optarg);
                break;
            case 'U':
                dbuser = pg_strdup(optarg);
                break;
            case 'w':
                dbgetpassword = -1;
                break;
            case 'W':
                dbgetpassword = 1;
                break;
/* replication options */
            case 'I':
                if (sscanf(optarg, "%X/%08X", &hi, &lo) != 2)
                    pg_fatal("could not parse start position \"%s\"", optarg);
                startpos = ((uint64) hi) << 32 | lo;
                break;
            case 'E':
                if (sscanf(optarg, "%X/%08X", &hi, &lo) != 2)
                    pg_fatal("could not parse end position \"%s\"", optarg);
                endpos = ((uint64) hi) << 32 | lo;
                break;
            case 'o':
                {
                    char       *data = pg_strdup(optarg);
                    char       *val = strchr(data, '=');
 
                    if (val != NULL)
                    {
                        /* remove =; separate data from val */
                        *val = '\0';
                        val++;
                    }
 
                    noptions += 1;
                    options = pg_realloc_array(options, char *, noptions * 2);
 
                    options[(noptions - 1) * 2] = data;
                    options[(noptions - 1) * 2 + 1] = val;
                }
 
                break;
            case 'P':
                plugin = pg_strdup(optarg);
                break;
            case 's':
                if (!option_parse_int(optarg, "-s/--status-interval", 0,
                                      INT_MAX / 1000,
                                      &standby_message_timeout))
                    exit(1);
                standby_message_timeout *= 1000;
                break;
            case 'S':
                replication_slot = pg_strdup(optarg);
                break;
/* action */
            case 1:
                do_create_slot = true;
                break;
            case 2:
                do_start_slot = true;
                break;
            case 3:
                do_drop_slot = true;
                break;
            case 4:
                slot_exists_ok = true;
                break;
 
            default:
                /* getopt_long already emitted a complaint */
                pg_log_error_hint("Try \"%s --help\" for more information.", progname);
                exit(1);
        }
    }
 
    /*
     * Any non-option arguments?
     */
    if (optind < argc)
    {
        pg_log_error("too many command-line arguments (first is \"%s\")",
                     argv[optind]);
        pg_log_error_hint("Try \"%s --help\" for more information.", progname);
        exit(1);
    }
 
    /*
     * Required arguments
     */
    if (replication_slot == NULL)
    {
        pg_log_error("no slot specified");
        pg_log_error_hint("Try \"%s --help\" for more information.", progname);
        exit(1);
    }
 
    if (do_start_slot && outfile == NULL)
    {
        pg_log_error("no target file specified");
        pg_log_error_hint("Try \"%s --help\" for more information.", progname);
        exit(1);
    }
 
    if (!do_drop_slot && dbname == NULL)
    {
        pg_log_error("no database specified");
        pg_log_error_hint("Try \"%s --help\" for more information.", progname);
        exit(1);
    }
 
    if (!do_drop_slot && !do_create_slot && !do_start_slot)
    {
        pg_log_error("at least one action needs to be specified");
        pg_log_error_hint("Try \"%s --help\" for more information.", progname);
        exit(1);
    }
 
    if (do_drop_slot && (do_create_slot || do_start_slot))
    {
        pg_log_error("cannot use --create-slot or --start together with --drop-slot");
        pg_log_error_hint("Try \"%s --help\" for more information.", progname);
        exit(1);
    }
 
    if (XLogRecPtrIsValid(startpos) && (do_create_slot || do_drop_slot))
    {
        pg_log_error("cannot use --create-slot or --drop-slot together with --startpos");
        pg_log_error_hint("Try \"%s --help\" for more information.", progname);
        exit(1);
    }
 
    if (XLogRecPtrIsValid(endpos) && !do_start_slot)
    {
        pg_log_error("--endpos may only be specified with --start");
        pg_log_error_hint("Try \"%s --help\" for more information.", progname);
        exit(1);
    }
 
    if (!do_create_slot)
    {
        if (two_phase)
        {
            pg_log_error("%s may only be specified with --create-slot", "--enable-two-phase");
            pg_log_error_hint("Try \"%s --help\" for more information.", progname);
            exit(1);
        }
 
        if (failover)
        {
            pg_log_error("%s may only be specified with --create-slot", "--enable-failover");
            pg_log_error_hint("Try \"%s --help\" for more information.", progname);
            exit(1);
        }
    }
 
    /*
     * Obtain a connection to server.  Notably, if we need a password, we want
     * to collect it from the user immediately.
     */
    conn = GetConnection();
    if (!conn)
        /* Error message already written in GetConnection() */
        exit(1);
    atexit(disconnect_atexit);
 
    /*
     * Trap signals.  (Don't do this until after the initial password prompt,
     * if one is needed, in GetConnection.)
     */
#ifndef WIN32
    pqsignal(SIGINT, sigexit_handler);
    pqsignal(SIGTERM, sigexit_handler);
    pqsignal(SIGHUP, sighup_handler);
#endif
 
    /*
     * Run IDENTIFY_SYSTEM to check the connection type for each action.
     * --create-slot and --start actions require a database-specific
     * replication connection because they handle logical replication slots.
     * --drop-slot can remove replication slots from any replication
     * connection without this restriction.
     */
    if (!RunIdentifySystem(conn, NULL, NULL, NULL, &db_name))
        exit(1);
 
    if (!do_drop_slot && db_name == NULL)
        pg_fatal("could not establish database-specific replication connection");
 
    /*
     * Set umask so that directories/files are created with the same
     * permissions as directories/files in the source data directory.
     *
     * pg_mode_mask is set to owner-only by default and then updated in
     * GetConnection() where we get the mode from the server-side with
     * RetrieveDataDirCreatePerm() and then call SetDataDirectoryCreatePerm().
     */
    umask(pg_mode_mask);
 
    /* Drop a replication slot. */
    if (do_drop_slot)
    {
        if (verbose)
            pg_log_info("dropping replication slot \"%s\"", replication_slot);
 
        if (!DropReplicationSlot(conn, replication_slot))
            exit(1);
    }
 
    /* Create a replication slot. */
    if (do_create_slot)
    {
        if (verbose)
            pg_log_info("creating replication slot \"%s\"", replication_slot);
 
        if (!CreateReplicationSlot(conn, replication_slot, plugin, false,
                                   false, false, slot_exists_ok, two_phase,
                                   failover))
            exit(1);
        startpos = InvalidXLogRecPtr;
    }
 
    if (!do_start_slot)
        exit(0);
 
    /* Stream loop */
    while (true)
    {
        StreamLogicalLog();
        if (time_to_abort)
        {
            /*
             * We've been Ctrl-C'ed or reached an exit limit condition. That's
             * not an error, so exit without an errorcode.
             */
            exit(0);
        }
 
        /*
         * Ensure all written data is flushed to disk before exiting or
         * starting a new replication.
         */
        if (outfd != -1)
            OutputFsync(feGetCurrentTimestamp());
 
        if (noloop)
        {
            pg_fatal("disconnected");
        }
        else
        {
            /* translator: check source for value for %d */
            pg_log_info("disconnected; waiting %d seconds to try again",
                        RECONNECT_SLEEP_TIME);
            pg_usleep(RECONNECT_SLEEP_TIME * 1000000);
        }
    }
}

References conn, CreateReplicationSlot(), data, dbgetpassword, dbhost, dbname, dbport, dbuser, disconnect_atexit(), do_create_slot, do_drop_slot, do_start_slot, DropReplicationSlot(), endpos, failover, fb(), feGetCurrentTimestamp(), fsync_interval, get_progname(), GetConnection(), getopt_long(), InvalidXLogRecPtr, no_argument, noloop, noptions, optarg, optind, option_parse_int(), outfd, outfile, OutputFsync(), pg_fatal, pg_log_error, pg_log_error_hint, pg_log_info, pg_logging_init(), pg_mode_mask, pg_realloc_array, pg_strdup(), PG_TEXTDOMAIN, pg_usleep(), plugin, pqsignal, progname, RECONNECT_SLEEP_TIME, replication_slot, required_argument, RunIdentifySystem(), set_pglocale_pgservice(), sigexit_handler(), SIGHUP, sighup_handler(), slot_exists_ok, standby_message_timeout, startpos, StreamLogicalLog(), time_to_abort, two_phase, usage(), val, verbose, and XLogRecPtrIsValid.

◆ OutputFsync()

static void OutputFsync ( TimestampTz now )

static

Definition at line 188 of file pg_recvlogical.c.

{
    output_last_fsync = now;
 
    output_fsync_lsn = output_written_lsn;
 
    /*
     * Save the last flushed position as the replication start point. On
     * reconnect, replication resumes from there to avoid re-sending flushed
     * data.
     */
    startpos = output_fsync_lsn;
 
    if (fsync_interval <= 0)
        return;
 
    if (!output_needs_fsync)
        return;
 
    output_needs_fsync = false;
 
    /* can only fsync if it's a regular file */
    if (!output_isfile)
        return;
 
    if (fsync(outfd) != 0)
        pg_fatal("could not fsync file \"%s\": %m", outfile);
}

References fsync, fsync_interval, now(), outfd, outfile, output_fsync_lsn, output_isfile, output_last_fsync, output_needs_fsync, output_written_lsn, pg_fatal, and startpos.

Referenced by flushAndSendFeedback(), main(), and StreamLogicalLog().

◆ prepareToTerminate()

static void prepareToTerminate	(	PGconn *	conn,
		XLogRecPtr	endpos,
		StreamStopReason	reason,
		XLogRecPtr	lsn
	)

static

Definition at line 1074 of file pg_recvlogical.c.

{
    /*
     * If pg_recvlogical is terminated by a signal, we can reach here without
     * sending final feedback. In that case, send feedback once more before
     * sending CopyDone so the replication slot can advance far enough to
     * reduce the chance of resending duplicate data when pg_recvlogical is
     * restarted.
     *
     * This is still only a best-effort attempt. Depending on when the signal
     * arrives, the receiver may have written decoded output that the server
     * cannot yet safely treat as confirmed, so a later restart can still see
     * duplicate data.
     *
     * For other termination cases, such as STREAM_STOP_KEEPALIVE and
     * STREAM_STOP_END_OF_WAL, feedback has already been sent before reaching
     * here, so there is no need to call flushAndSendFeedback() again.
     */
    if (reason == STREAM_STOP_SIGNAL)
    {
        TimestampTz now = feGetCurrentTimestamp();
 
        (void) flushAndSendFeedback(conn, &now);
    }
 
    (void) PQputCopyEnd(conn, NULL);
    (void) PQflush(conn);
 
    if (verbose)
    {
        switch (reason)
        {
            case STREAM_STOP_SIGNAL:
                pg_log_info("received interrupt signal, exiting");
                break;
            case STREAM_STOP_KEEPALIVE:
                pg_log_info("end position %X/%08X reached by keepalive",
                            LSN_FORMAT_ARGS(endpos));
                break;
            case STREAM_STOP_END_OF_WAL:
                Assert(XLogRecPtrIsValid(lsn));
                pg_log_info("end position %X/%08X reached by WAL record at %X/%08X",
                            LSN_FORMAT_ARGS(endpos), LSN_FORMAT_ARGS(lsn));
                break;
            case STREAM_STOP_NONE:
                Assert(false);
                break;
        }
    }
}

References Assert, conn, endpos, fb(), feGetCurrentTimestamp(), flushAndSendFeedback(), LSN_FORMAT_ARGS, now(), pg_log_info, PQflush(), PQputCopyEnd(), STREAM_STOP_END_OF_WAL, STREAM_STOP_KEEPALIVE, STREAM_STOP_NONE, STREAM_STOP_SIGNAL, verbose, and XLogRecPtrIsValid.

Referenced by StreamLogicalLog().

◆ sendFeedback()

static bool sendFeedback	(	PGconn *	conn,
		TimestampTz	now,
		bool	force,
		bool	replyRequested
	)

static

Definition at line 129 of file pg_recvlogical.c.

{
    static XLogRecPtr last_written_lsn = InvalidXLogRecPtr;
    static XLogRecPtr last_fsync_lsn = InvalidXLogRecPtr;
 
    char        replybuf[1 + 8 + 8 + 8 + 8 + 1];
    int         len = 0;
 
    /*
     * we normally don't want to send superfluous feedback, but if it's
     * because of a timeout we need to, otherwise wal_sender_timeout will kill
     * us.
     */
    if (!force &&
        last_written_lsn == output_written_lsn &&
        last_fsync_lsn == output_fsync_lsn)
        return true;
 
    if (verbose)
        pg_log_info("confirming write up to %X/%08X, flush to %X/%08X (slot %s)",
                    LSN_FORMAT_ARGS(output_written_lsn),
                    LSN_FORMAT_ARGS(output_fsync_lsn),
                    replication_slot);
 
    replybuf[len] = PqReplMsg_StandbyStatusUpdate;
    len += 1;
    fe_sendint64(output_written_lsn, &replybuf[len]);   /* write */
    len += 8;
    fe_sendint64(output_fsync_lsn, &replybuf[len]); /* flush */
    len += 8;
    fe_sendint64(InvalidXLogRecPtr, &replybuf[len]);    /* apply */
    len += 8;
    fe_sendint64(now, &replybuf[len]);  /* sendTime */
    len += 8;
    replybuf[len] = replyRequested ? 1 : 0; /* replyRequested */
    len += 1;
 
    startpos = output_written_lsn;
    last_written_lsn = output_written_lsn;
    last_fsync_lsn = output_fsync_lsn;
 
    if (PQputCopyData(conn, replybuf, len) <= 0 || PQflush(conn))
    {
        pg_log_error("could not send feedback packet: %s",
                     PQerrorMessage(conn));
        return false;
    }
 
    return true;
}

References conn, fb(), fe_sendint64(), InvalidXLogRecPtr, len, LSN_FORMAT_ARGS, now(), output_fsync_lsn, output_written_lsn, pg_log_error, pg_log_info, PQerrorMessage(), PQflush(), PQputCopyData(), PqReplMsg_StandbyStatusUpdate, replication_slot, startpos, and verbose.

Referenced by flushAndSendFeedback(), and StreamLogicalLog().

◆ sigexit_handler()

static void sigexit_handler ( SIGNAL_ARGS )

static

Definition at line 678 of file pg_recvlogical.c.

{
    stop_reason = STREAM_STOP_SIGNAL;
    time_to_abort = true;
}

References stop_reason, STREAM_STOP_SIGNAL, and time_to_abort.

Referenced by main().

◆ sighup_handler()

static void sighup_handler ( SIGNAL_ARGS )

static

Definition at line 688 of file pg_recvlogical.c.

{
    output_reopen = true;
}

References output_reopen.

Referenced by main().

◆ StreamLogicalLog()

static void StreamLogicalLog ( void )

static

Definition at line 221 of file pg_recvlogical.c.

{
    PGresult   *res;
    char       *copybuf = NULL;
    TimestampTz last_status = -1;
    PQExpBuffer query;
    XLogRecPtr  cur_record_lsn;
 
    cur_record_lsn = InvalidXLogRecPtr;
 
    /*
     * Connect in replication mode to the server
     */
    if (!conn)
        conn = GetConnection();
    if (!conn)
        /* Error message already written in GetConnection() */
        return;
 
    /*
     * Start the replication
     */
    if (verbose)
        pg_log_info("starting log streaming at %X/%08X (slot %s)",
                    LSN_FORMAT_ARGS(startpos),
                    replication_slot);
 
    /* Initiate the replication stream at specified location */
    query = createPQExpBuffer();
    appendPQExpBufferStr(query, "START_REPLICATION SLOT ");
    AppendQuotedIdentifier(query, replication_slot);
    appendPQExpBuffer(query, " LOGICAL %X/%08X", LSN_FORMAT_ARGS(startpos));
 
    /* print options if there are any */
    if (noptions)
        appendPQExpBufferStr(query, " (");
 
    for (size_t i = 0; i < noptions; i++)
    {
        /* separator */
        if (i > 0)
            appendPQExpBufferStr(query, ", ");
 
        /* write option name */
        AppendQuotedIdentifier(query, options[i * 2]);
 
        /* write option value if specified */
        if (options[i * 2 + 1] != NULL)
        {
            appendPQExpBufferChar(query, ' ');
            AppendQuotedLiteral(query, options[i * 2 + 1]);
        }
    }
 
    if (noptions)
        appendPQExpBufferChar(query, ')');
 
    res = PQexec(conn, query->data);
    if (PQresultStatus(res) != PGRES_COPY_BOTH)
    {
        pg_log_error("could not send replication command \"%s\": %s",
                     query->data, PQresultErrorMessage(res));
        PQclear(res);
        goto error;
    }
    PQclear(res);
    resetPQExpBuffer(query);
 
    if (verbose)
        pg_log_info("streaming initiated");
 
    while (!time_to_abort)
    {
        int         r;
        size_t      bytes_left;
        size_t      bytes_written;
        TimestampTz now;
        size_t      hdr_len;
 
        cur_record_lsn = InvalidXLogRecPtr;
 
        if (copybuf != NULL)
        {
            PQfreemem(copybuf);
            copybuf = NULL;
        }
 
        /*
         * Potentially send a status message to the primary.
         */
        now = feGetCurrentTimestamp();
 
        if (outfd != -1 &&
            feTimestampDifferenceExceeds(output_last_fsync, now,
                                         fsync_interval))
            OutputFsync(now);
 
        if (standby_message_timeout > 0 &&
            feTimestampDifferenceExceeds(last_status, now,
                                         standby_message_timeout))
        {
            /* Time to send feedback! */
            if (!sendFeedback(conn, now, true, false))
                goto error;
 
            last_status = now;
        }
 
        /* got SIGHUP, close output file */
        if (outfd != -1 && output_reopen && strcmp(outfile, "-") != 0)
        {
            now = feGetCurrentTimestamp();
            OutputFsync(now);
            close(outfd);
            outfd = -1;
        }
        output_reopen = false;
 
        /* open the output file, if not open yet */
        if (outfd == -1)
        {
            struct stat statbuf;
 
            if (strcmp(outfile, "-") == 0)
                outfd = fileno(stdout);
            else
                outfd = open(outfile, O_CREAT | O_APPEND | O_WRONLY | PG_BINARY,
                             pg_file_create_mode);
            if (outfd == -1)
            {
                pg_log_error("could not open log file \"%s\": %m", outfile);
                goto error;
            }
 
            if (fstat(outfd, &statbuf) != 0)
            {
                pg_log_error("could not stat file \"%s\": %m", outfile);
                goto error;
            }
 
            output_isfile = S_ISREG(statbuf.st_mode) && !isatty(outfd);
        }
 
        r = PQgetCopyData(conn, &copybuf, 1);
        if (r == 0)
        {
            /*
             * In async mode, and no data available. We block on reading but
             * not more than the specified timeout, so that we can send a
             * response back to the client.
             */
            fd_set      input_mask;
            TimestampTz message_target = 0;
            TimestampTz fsync_target = 0;
            struct timeval timeout;
            struct timeval *timeoutptr = NULL;
 
            if (PQsocket(conn) < 0)
            {
                pg_log_error("invalid socket: %s", PQerrorMessage(conn));
                goto error;
            }
 
            FD_ZERO(&input_mask);
            FD_SET(PQsocket(conn), &input_mask);
 
            /* Compute when we need to wakeup to send a keepalive message. */
            if (standby_message_timeout)
                message_target = last_status + (standby_message_timeout - 1) *
                    ((int64) 1000);
 
            /* Compute when we need to wakeup to fsync the output file. */
            if (fsync_interval > 0 && output_needs_fsync)
                fsync_target = output_last_fsync + (fsync_interval - 1) *
                    ((int64) 1000);
 
            /* Now compute when to wakeup. */
            if (message_target > 0 || fsync_target > 0)
            {
                TimestampTz targettime;
                long        secs;
                int         usecs;
 
                targettime = message_target;
 
                if (fsync_target > 0 && fsync_target < targettime)
                    targettime = fsync_target;
 
                feTimestampDifference(now,
                                      targettime,
                                      &secs,
                                      &usecs);
                if (secs <= 0)
                    timeout.tv_sec = 1; /* Always sleep at least 1 sec */
                else
                    timeout.tv_sec = secs;
                timeout.tv_usec = usecs;
                timeoutptr = &timeout;
            }
 
            r = select(PQsocket(conn) + 1, &input_mask, NULL, NULL, timeoutptr);
            if (r == 0 || (r < 0 && errno == EINTR))
            {
                /*
                 * Got a timeout or signal. Continue the loop and either
                 * deliver a status packet to the server or just go back into
                 * blocking.
                 */
                continue;
            }
            else if (r < 0)
            {
                pg_log_error("%s() failed: %m", "select");
                goto error;
            }
 
            /* Else there is actually data on the socket */
            if (PQconsumeInput(conn) == 0)
            {
                pg_log_error("could not receive data from WAL stream: %s",
                             PQerrorMessage(conn));
                goto error;
            }
            continue;
        }
 
        /* End of copy stream */
        if (r == -1)
            break;
 
        /* Failure while reading the copy stream */
        if (r == -2)
        {
            pg_log_error("could not read COPY data: %s",
                         PQerrorMessage(conn));
            goto error;
        }
 
        /* Check the message type. */
        if (copybuf[0] == PqReplMsg_Keepalive)
        {
            int         pos;
            bool        replyRequested;
            XLogRecPtr  walEnd;
            bool        endposReached = false;
 
            /*
             * Parse the keepalive message, enclosed in the CopyData message.
             * We just check if the server requested a reply, and ignore the
             * rest.
             */
            pos = 1;            /* skip msgtype PqReplMsg_Keepalive */
            walEnd = fe_recvint64(&copybuf[pos]);
            output_written_lsn = Max(walEnd, output_written_lsn);
 
            pos += 8;           /* read walEnd */
 
            pos += 8;           /* skip sendTime */
 
            if (r < pos + 1)
            {
                pg_log_error("streaming header too small: %d", r);
                goto error;
            }
            replyRequested = copybuf[pos];
 
            if (XLogRecPtrIsValid(endpos) && walEnd >= endpos)
            {
                /*
                 * If there's nothing to read on the socket until a keepalive
                 * we know that the server has nothing to send us; and if
                 * walEnd has passed endpos, we know nothing else can have
                 * committed before endpos.  So we can bail out now.
                 */
                endposReached = true;
            }
 
            /* Send a reply, if necessary */
            if (replyRequested || endposReached)
            {
                if (!flushAndSendFeedback(conn, &now))
                    goto error;
                last_status = now;
            }
 
            if (endposReached)
            {
                stop_reason = STREAM_STOP_KEEPALIVE;
                time_to_abort = true;
                break;
            }
 
            continue;
        }
        else if (copybuf[0] != PqReplMsg_WALData)
        {
            pg_log_error("unrecognized streaming header: \"%c\"",
                         copybuf[0]);
            goto error;
        }
 
        /*
         * Read the header of the WALData message, enclosed in the CopyData
         * message. We only need the WAL location field (dataStart), the rest
         * of the header is ignored.
         */
        hdr_len = 1;            /* msgtype PqReplMsg_WALData */
        hdr_len += 8;           /* dataStart */
        hdr_len += 8;           /* walEnd */
        hdr_len += 8;           /* sendTime */
        if (r < hdr_len + 1)
        {
            pg_log_error("streaming header too small: %d", r);
            goto error;
        }
 
        /* Extract WAL location for this block */
        cur_record_lsn = fe_recvint64(&copybuf[1]);
 
        if (XLogRecPtrIsValid(endpos) && cur_record_lsn > endpos)
        {
            /*
             * We've read past our endpoint, so prepare to go away being
             * cautious about what happens to our output data.
             */
            if (!flushAndSendFeedback(conn, &now))
                goto error;
            stop_reason = STREAM_STOP_END_OF_WAL;
            time_to_abort = true;
            break;
        }
 
        output_written_lsn = Max(cur_record_lsn, output_written_lsn);
 
        bytes_left = r - hdr_len;
        bytes_written = 0;
 
        /* signal that a fsync is needed */
        output_needs_fsync = true;
 
        while (bytes_left)
        {
            ssize_t     ret;
 
            ret = write(outfd,
                        copybuf + hdr_len + bytes_written,
                        bytes_left);
 
            if (ret < 0)
            {
                pg_log_error("could not write %zu bytes to log file \"%s\": %m",
                             bytes_left, outfile);
                goto error;
            }
 
            /* Write was successful, advance our position */
            bytes_written += ret;
            bytes_left -= ret;
        }
 
        if (write(outfd, "\n", 1) != 1)
        {
            pg_log_error("could not write %zu bytes to log file \"%s\": %m",
                         (size_t) 1, outfile);
            goto error;
        }
 
        if (XLogRecPtrIsValid(endpos) && cur_record_lsn == endpos)
        {
            /* endpos was exactly the record we just processed, we're done */
            if (!flushAndSendFeedback(conn, &now))
                goto error;
            stop_reason = STREAM_STOP_END_OF_WAL;
            time_to_abort = true;
            break;
        }
    }
 
    /* Clean up connection state if stream has been aborted */
    if (time_to_abort)
        prepareToTerminate(conn, endpos, stop_reason, cur_record_lsn);
 
    res = PQgetResult(conn);
    if (PQresultStatus(res) == PGRES_COPY_OUT)
    {
        PQclear(res);
 
        /*
         * We're doing a client-initiated clean exit and have sent CopyDone to
         * the server. Drain any messages, so we don't miss a last-minute
         * ErrorResponse. The walsender stops generating WALData records once
         * it sees CopyDone, so expect this to finish quickly. After CopyDone,
         * it's too late for sendFeedback(), even if this were to take a long
         * time. Hence, use synchronous-mode PQgetCopyData().
         */
        while (1)
        {
            int         r;
 
            if (copybuf != NULL)
            {
                PQfreemem(copybuf);
                copybuf = NULL;
            }
            r = PQgetCopyData(conn, &copybuf, 0);
            if (r == -1)
                break;
            if (r == -2)
            {
                pg_log_error("could not read COPY data: %s",
                             PQerrorMessage(conn));
                time_to_abort = false;  /* unclean exit */
                goto error;
            }
        }
 
        res = PQgetResult(conn);
    }
    if (PQresultStatus(res) != PGRES_COMMAND_OK)
    {
        pg_log_error("unexpected termination of replication stream: %s",
                     PQresultErrorMessage(res));
        PQclear(res);
        goto error;
    }
    PQclear(res);
 
    if (outfd != -1 && strcmp(outfile, "-") != 0)
    {
        TimestampTz t = feGetCurrentTimestamp();
 
        OutputFsync(t);
        if (close(outfd) != 0)
            pg_log_error("could not close file \"%s\": %m", outfile);
    }
    outfd = -1;
error:
    if (copybuf != NULL)
    {
        PQfreemem(copybuf);
        copybuf = NULL;
    }
    destroyPQExpBuffer(query);
    PQfinish(conn);
    conn = NULL;
}

References appendPQExpBuffer(), appendPQExpBufferChar(), appendPQExpBufferStr(), AppendQuotedIdentifier, AppendQuotedLiteral, close, conn, copybuf, createPQExpBuffer(), PQExpBufferData::data, destroyPQExpBuffer(), EINTR, endpos, error(), fb(), fe_recvint64(), feGetCurrentTimestamp(), feTimestampDifference(), feTimestampDifferenceExceeds(), flushAndSendFeedback(), fstat, fsync_interval, GetConnection(), i, InvalidXLogRecPtr, LSN_FORMAT_ARGS, Max, noptions, now(), outfd, outfile, output_isfile, output_last_fsync, output_needs_fsync, output_reopen, output_written_lsn, OutputFsync(), PG_BINARY, pg_file_create_mode, pg_log_error, pg_log_info, PGRES_COMMAND_OK, PGRES_COPY_BOTH, PGRES_COPY_OUT, PQclear, PQconsumeInput(), PQerrorMessage(), PQexec(), PQfinish(), PQfreemem(), PQgetCopyData(), PQgetResult, PqReplMsg_Keepalive, PqReplMsg_WALData, PQresultErrorMessage, PQresultStatus, PQsocket(), prepareToTerminate(), replication_slot, resetPQExpBuffer(), S_ISREG, select, sendFeedback(), standby_message_timeout, startpos, stop_reason, STREAM_STOP_END_OF_WAL, STREAM_STOP_KEEPALIVE, time_to_abort, verbose, write, and XLogRecPtrIsValid.

Referenced by main().

◆ usage()

static void usage ( void )

static

Definition at line 82 of file pg_recvlogical.c.

{
    printf(_("%s controls PostgreSQL logical decoding streams.\n\n"),
           progname);
    printf(_("Usage:\n"));
    printf(_("  %s [OPTION]...\n"), progname);
    printf(_("\nAction to be performed:\n"));
    printf(_("      --create-slot      create a new replication slot (for the slot's name see --slot)\n"));
    printf(_("      --drop-slot        drop the replication slot (for the slot's name see --slot)\n"));
    printf(_("      --start            start streaming in a replication slot (for the slot's name see --slot)\n"));
    printf(_("\nOptions:\n"));
    printf(_("      --enable-failover  enable replication slot synchronization to standby servers when\n"
             "                         creating a replication slot\n"));
    printf(_("  -E, --endpos=LSN       exit after receiving the specified LSN\n"));
    printf(_("  -f, --file=FILE        receive log into this file, - for stdout\n"));
    printf(_("  -F  --fsync-interval=SECS\n"
             "                         time between fsyncs to the output file (default: %d)\n"), (fsync_interval / 1000));
    printf(_("      --if-not-exists    do not error if slot already exists when creating a slot\n"));
    printf(_("  -I, --startpos=LSN     where in an existing slot should the streaming start\n"));
    printf(_("  -n, --no-loop          do not loop on connection lost\n"));
    printf(_("  -o, --option=NAME[=VALUE]\n"
             "                         pass option NAME with optional value VALUE to the\n"
             "                         output plugin\n"));
    printf(_("  -P, --plugin=PLUGIN    use output plugin PLUGIN (default: %s)\n"), plugin);
    printf(_("  -s, --status-interval=SECS\n"
             "                         time between status packets sent to server (default: %d)\n"), (standby_message_timeout / 1000));
    printf(_("  -S, --slot=SLOTNAME    name of the logical replication slot\n"));
    printf(_("  -t, --enable-two-phase enable decoding of prepared transactions when creating a slot\n"));
    printf(_("      --two-phase        (same as --enable-two-phase, deprecated)\n"));
    printf(_("  -v, --verbose          output verbose messages\n"));
    printf(_("  -V, --version          output version information, then exit\n"));
    printf(_("  -?, --help             show this help, then exit\n"));
    printf(_("\nConnection options:\n"));
    printf(_("  -d, --dbname=DBNAME    database to connect to\n"));
    printf(_("  -h, --host=HOSTNAME    database server host or socket directory\n"));
    printf(_("  -p, --port=PORT        database server port number\n"));
    printf(_("  -U, --username=NAME    connect as specified database user\n"));
    printf(_("  -w, --no-password      never prompt for password\n"));
    printf(_("  -W, --password         force password prompt (should happen automatically)\n"));
    printf(_("\nReport bugs to <%s>.\n"), PACKAGE_BUGREPORT);
    printf(_("%s home page: <%s>\n"), PACKAGE_NAME, PACKAGE_URL);
}