libpqwalreceiver.c File Reference

#include "postgres.h"
#include <unistd.h>
#include <sys/time.h>
#include "common/connect.h"
#include "funcapi.h"
#include "libpq-fe.h"
#include "libpq/libpq-be-fe-helpers.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "pqexpbuffer.h"
#include "replication/walreceiver.h"
#include "storage/latch.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
#include "utils/pg_lsn.h"
#include "utils/tuplestore.h"

Include dependency graph for libpqwalreceiver.c:

Go to the source code of this file.

Data Structures
struct	WalReceiverConn

Macros
#define	appendQuotedIdentifier(b, s)   appendQuotedString(b, s, '"')
#define	appendQuotedLiteral(b, s)   appendQuotedString(b, s, '\'')

Functions
	PG_MODULE_MAGIC_EXT (.name="libpqwalreceiver",.version=PG_VERSION)
static WalReceiverConn *	libpqrcv_connect (const char *conninfo, bool replication, bool logical, bool must_use_password, const char *appname, char **err)
static void	libpqrcv_check_conninfo (const char *conninfo, bool must_use_password)
static char *	libpqrcv_get_conninfo (WalReceiverConn *conn)
static void	libpqrcv_get_senderinfo (WalReceiverConn *conn, char **sender_host, int *sender_port)
static char *	libpqrcv_identify_system (WalReceiverConn *conn, TimeLineID *primary_tli)
static char *	libpqrcv_get_dbname_from_conninfo (const char *connInfo)
static char *	libpqrcv_get_option_from_conninfo (const char *connInfo, const char *keyword)
static int	libpqrcv_server_version (WalReceiverConn *conn)
static void	libpqrcv_readtimelinehistoryfile (WalReceiverConn *conn, TimeLineID tli, char **filename, char **content, int *len)
static bool	libpqrcv_startstreaming (WalReceiverConn *conn, const WalRcvStreamOptions *options)
static void	libpqrcv_endstreaming (WalReceiverConn *conn, TimeLineID *next_tli)
static int	libpqrcv_receive (WalReceiverConn *conn, char **buffer, pgsocket *wait_fd)
static void	libpqrcv_send (WalReceiverConn *conn, const char *buffer, int nbytes)
static char *	libpqrcv_create_slot (WalReceiverConn *conn, const char *slotname, bool temporary, bool two_phase, bool failover, CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)
static void	libpqrcv_alter_slot (WalReceiverConn *conn, const char *slotname, const bool *failover, const bool *two_phase)
static pid_t	libpqrcv_get_backend_pid (WalReceiverConn *conn)
static WalRcvExecResult *	libpqrcv_exec (WalReceiverConn *conn, const char *query, const int nRetTypes, const Oid *retTypes)
static void	libpqrcv_disconnect (WalReceiverConn *conn)
static char *	stringlist_to_identifierstr (List *strings)
void	_PG_init (void)
static void	appendQuotedString (StringInfo buf, const char *str, char quote)
static void	libpqrcv_processTuples (PGresult *pgres, WalRcvExecResult *walres, const int nRetTypes, const Oid *retTypes)

Variables
static WalReceiverFunctionsType	PQWalReceiverFunctions

Macro Definition Documentation

appendQuotedIdentifier

#define appendQuotedIdentifier	(		b,
			s
	)		appendQuotedString(b, s, '"')

Definition at line 548 of file libpqwalreceiver.c.

appendQuotedLiteral

#define appendQuotedLiteral	(		b,
			s
	)		appendQuotedString(b, s, '\'')

Definition at line 549 of file libpqwalreceiver.c.

Function Documentation

_PG_init()

void _PG_init

(

void

)

Definition at line 125 of file libpqwalreceiver.c.

{
    if (WalReceiverFunctions != NULL)
        elog(ERROR, "libpqwalreceiver already loaded");
    WalReceiverFunctions = &PQWalReceiverFunctions;
}

References elog, ERROR, fb(), PQWalReceiverFunctions, and WalReceiverFunctions.

appendQuotedString()

static void appendQuotedString ( StringInfo  buf, const char *  str, char  quote  )

static

Definition at line 534 of file libpqwalreceiver.c.

{
    appendStringInfoChar(buf, quote);
    while (*str)
    {
        char        c = *str++;
 
        if (c == quote)
            appendStringInfoChar(buf, c);
        appendStringInfoChar(buf, c);
    }
    appendStringInfoChar(buf, quote);
}

References appendStringInfoChar(), buf, and str.

libpqrcv_alter_slot()

static void libpqrcv_alter_slot ( WalReceiverConn *  conn, const char *  slotname, const bool *  failover, const bool *  two_phase  )

static

Definition at line 1021 of file libpqwalreceiver.c.

{
    StringInfoData cmd;
    PGresult   *res;
 
    initStringInfo(&cmd);
    appendStringInfoString(&cmd, "ALTER_REPLICATION_SLOT ");
    appendQuotedIdentifier(&cmd, slotname);
    appendStringInfoString(&cmd, " ( ");
 
    if (failover)
        appendStringInfo(&cmd, "FAILOVER %s",
                         *failover ? "true" : "false");
 
    if (failover && two_phase)
        appendStringInfoString(&cmd, ", ");
 
    if (two_phase)
        appendStringInfo(&cmd, "TWO_PHASE %s",
                         *two_phase ? "true" : "false");
 
    appendStringInfoString(&cmd, " );");
 
    res = libpqsrv_exec(conn->streamConn, cmd.data,
                        WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
    pfree(cmd.data);
 
    if (PQresultStatus(res) != PGRES_COMMAND_OK)
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg("could not alter replication slot \"%s\": %s",
                        slotname, pchomp(PQerrorMessage(conn->streamConn)))));
 
    PQclear(res);
}

References appendQuotedIdentifier, appendStringInfo(), appendStringInfoString(), conn, StringInfoData::data, ereport, errcode(), ERRCODE_PROTOCOL_VIOLATION, errmsg, ERROR, failover, fb(), initStringInfo(), libpqsrv_exec(), pchomp(), pfree(), PGRES_COMMAND_OK, PQclear, PQerrorMessage(), PQresultStatus, and two_phase.

libpqrcv_check_conninfo()

static void libpqrcv_check_conninfo ( const char *  conninfo, bool  must_use_password  )

static

Definition at line 296 of file libpqwalreceiver.c.

{
    PQconninfoOption *opts = NULL;
    PQconninfoOption *opt;
    char       *err = NULL;
 
    opts = PQconninfoParse(conninfo, &err);
    if (opts == NULL)
    {
        /* The error string is malloc'd, so we must free it explicitly */
        char       *errcopy = err ? pstrdup(err) : "out of memory";
 
        PQfreemem(err);
        ereport(ERROR,
                (errcode(ERRCODE_SYNTAX_ERROR),
                 errmsg("invalid connection string syntax: %s", errcopy)));
    }
 
    if (must_use_password)
    {
        bool        uses_password = false;
 
        for (opt = opts; opt->keyword != NULL; ++opt)
        {
            /* Ignore connection options that are not present. */
            if (opt->val == NULL)
                continue;
 
            if (strcmp(opt->keyword, "password") == 0 && opt->val[0] != '\0')
            {
                uses_password = true;
                break;
            }
        }
 
        if (!uses_password)
        {
            /* malloc'd, so we must free it explicitly */
            PQconninfoFree(opts);
 
            ereport(ERROR,
                    (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
                     errmsg("password is required"),
                     errdetail("Non-superusers must provide a password in the connection string.")));
        }
    }
 
    PQconninfoFree(opts);
}

References ereport, err(), errcode(), errdetail(), errmsg, ERROR, fb(), _PQconninfoOption::keyword, opts, PQconninfoFree(), PQconninfoParse(), PQfreemem(), pstrdup(), and _PQconninfoOption::val.

Referenced by libpqrcv_connect().

libpqrcv_connect()

static WalReceiverConn * libpqrcv_connect ( const char *  conninfo, bool  replication, bool  logical, bool  must_use_password, const char *  appname, char **  err  )

static

Definition at line 148 of file libpqwalreceiver.c.

{
    WalReceiverConn *conn;
    const char *keys[6];
    const char *vals[6];
    int         i = 0;
    char       *options_val = NULL;
 
    /*
     * Re-validate connection string. The validation already happened at DDL
     * time, but the subscription owner may have changed. If we don't recheck
     * with the correct must_use_password, it's possible that the connection
     * will obtain the password from a different source, such as PGPASSFILE or
     * PGPASSWORD.
     */
    libpqrcv_check_conninfo(conninfo, must_use_password);
 
    /*
     * We use the expand_dbname parameter to process the connection string (or
     * URI), and pass some extra options.
     */
    keys[i] = "dbname";
    vals[i] = conninfo;
 
    /* We can not have logical without replication */
    Assert(replication || !logical);
 
    if (replication)
    {
        keys[++i] = "replication";
        vals[i] = logical ? "database" : "true";
 
        if (logical)
        {
            char       *opt = NULL;
 
            /* Tell the publisher to translate to our encoding */
            keys[++i] = "client_encoding";
            vals[i] = GetDatabaseEncodingName();
 
            /*
             * Force assorted GUC parameters to settings that ensure that the
             * publisher will output data values in a form that is unambiguous
             * to the subscriber.  (We don't want to modify the subscriber's
             * GUC settings, since that might surprise user-defined code
             * running in the subscriber, such as triggers.)  This should
             * match what pg_dump does.
             */
            opt = libpqrcv_get_option_from_conninfo(conninfo, "options");
            options_val = psprintf("%s -c datestyle=ISO -c intervalstyle=postgres -c extra_float_digits=3",
                                   (opt == NULL) ? "" : opt);
            keys[++i] = "options";
            vals[i] = options_val;
            if (opt != NULL)
                pfree(opt);
        }
        else
        {
            /*
             * The database name is ignored by the server in replication mode,
             * but specify "replication" for .pgpass lookup.
             */
            keys[++i] = "dbname";
            vals[i] = "replication";
        }
    }
 
    keys[++i] = "fallback_application_name";
    vals[i] = appname;
 
    keys[++i] = NULL;
    vals[i] = NULL;
 
    Assert(i < lengthof(keys));
 
    conn = palloc0_object(WalReceiverConn);
    conn->streamConn =
        libpqsrv_connect_params_start(keys, vals,
                                       /* expand_dbname = */ true);
    PQsetNoticeReceiver(conn->streamConn, libpqsrv_notice_receiver,
                        "received message via replication");
    libpqsrv_connect_complete(conn->streamConn,
                              WAIT_EVENT_LIBPQWALRECEIVER_CONNECT);
 
    if (options_val != NULL)
        pfree(options_val);
 
    if (PQstatus(conn->streamConn) != CONNECTION_OK)
        goto bad_connection_errmsg;
 
    if (must_use_password && !PQconnectionUsedPassword(conn->streamConn))
    {
        libpqsrv_disconnect(conn->streamConn);
        pfree(conn);
 
        ereport(ERROR,
                (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
                 errmsg("password is required"),
                 errdetail("Non-superuser cannot connect if the server does not request a password."),
                 errhint("Target server's authentication method must be changed, or set password_required=false in the subscription parameters.")));
    }
 
    /*
     * Set always-secure search path for the cases where the connection is
     * used to run SQL queries, so malicious users can't get control.
     */
    if (!replication || logical)
    {
        PGresult   *res;
 
        res = libpqsrv_exec(conn->streamConn,
                            ALWAYS_SECURE_SEARCH_PATH_SQL,
                            WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
        if (PQresultStatus(res) != PGRES_TUPLES_OK)
        {
            PQclear(res);
            *err = psprintf(_("could not clear search path: %s"),
                            pchomp(PQerrorMessage(conn->streamConn)));
            goto bad_connection;
        }
        PQclear(res);
    }
 
    conn->logical = logical;
 
    return conn;
 
    /* error path, using libpq's error message */
bad_connection_errmsg:
    *err = pchomp(PQerrorMessage(conn->streamConn));
 
    /* error path, error already set */
bad_connection:
    libpqsrv_disconnect(conn->streamConn);
    pfree(conn);
    return NULL;
}

References _, ALWAYS_SECURE_SEARCH_PATH_SQL, Assert, conn, CONNECTION_OK, ereport, err(), errcode(), errdetail(), errhint(), errmsg, ERROR, fb(), GetDatabaseEncodingName(), i, lengthof, libpqrcv_check_conninfo(), libpqrcv_get_option_from_conninfo(), libpqsrv_connect_complete(), libpqsrv_connect_params_start(), libpqsrv_disconnect(), libpqsrv_exec(), libpqsrv_notice_receiver(), palloc0_object, pchomp(), pfree(), PGRES_TUPLES_OK, PQclear, PQconnectionUsedPassword(), PQerrorMessage(), PQresultStatus, PQsetNoticeReceiver(), PQstatus(), and psprintf().

libpqrcv_create_slot()

static char * libpqrcv_create_slot ( WalReceiverConn *  conn, const char *  slotname, bool  temporary, bool  two_phase, bool  failover, CRSSnapshotAction  snapshot_action, XLogRecPtr *  lsn  )

static

Definition at line 908 of file libpqwalreceiver.c.

{
    PGresult   *res;
    StringInfoData cmd;
    char       *snapshot;
    int         use_new_options_syntax;
 
    use_new_options_syntax = (PQserverVersion(conn->streamConn) >= 150000);
 
    initStringInfo(&cmd);
 
    appendStringInfoString(&cmd, "CREATE_REPLICATION_SLOT ");
    appendQuotedIdentifier(&cmd, slotname);
 
    if (temporary)
        appendStringInfoString(&cmd, " TEMPORARY");
 
    if (conn->logical)
    {
        appendStringInfoString(&cmd, " LOGICAL pgoutput ");
        if (use_new_options_syntax)
            appendStringInfoChar(&cmd, '(');
        if (two_phase)
        {
            appendStringInfoString(&cmd, "TWO_PHASE");
            if (use_new_options_syntax)
                appendStringInfoString(&cmd, ", ");
            else
                appendStringInfoChar(&cmd, ' ');
        }
 
        if (failover)
        {
            appendStringInfoString(&cmd, "FAILOVER");
            if (use_new_options_syntax)
                appendStringInfoString(&cmd, ", ");
            else
                appendStringInfoChar(&cmd, ' ');
        }
 
        if (use_new_options_syntax)
        {
            switch (snapshot_action)
            {
                case CRS_EXPORT_SNAPSHOT:
                    appendStringInfoString(&cmd, "SNAPSHOT 'export'");
                    break;
                case CRS_NOEXPORT_SNAPSHOT:
                    appendStringInfoString(&cmd, "SNAPSHOT 'nothing'");
                    break;
                case CRS_USE_SNAPSHOT:
                    appendStringInfoString(&cmd, "SNAPSHOT 'use'");
                    break;
            }
        }
        else
        {
            switch (snapshot_action)
            {
                case CRS_EXPORT_SNAPSHOT:
                    appendStringInfoString(&cmd, "EXPORT_SNAPSHOT");
                    break;
                case CRS_NOEXPORT_SNAPSHOT:
                    appendStringInfoString(&cmd, "NOEXPORT_SNAPSHOT");
                    break;
                case CRS_USE_SNAPSHOT:
                    appendStringInfoString(&cmd, "USE_SNAPSHOT");
                    break;
            }
        }
 
        if (use_new_options_syntax)
            appendStringInfoChar(&cmd, ')');
    }
    else
    {
        if (use_new_options_syntax)
            appendStringInfoString(&cmd, " PHYSICAL (RESERVE_WAL)");
        else
            appendStringInfoString(&cmd, " PHYSICAL RESERVE_WAL");
    }
 
    res = libpqsrv_exec(conn->streamConn,
                        cmd.data,
                        WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
    pfree(cmd.data);
 
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg("could not create replication slot \"%s\": %s",
                        slotname, pchomp(PQerrorMessage(conn->streamConn)))));
 
    if (lsn)
        *lsn = DatumGetLSN(DirectFunctionCall1Coll(pg_lsn_in, InvalidOid,
                                                   CStringGetDatum(PQgetvalue(res, 0, 1))));
 
    if (!PQgetisnull(res, 0, 2))
        snapshot = pstrdup(PQgetvalue(res, 0, 2));
    else
        snapshot = NULL;
 
    PQclear(res);
 
    return snapshot;
}

libpqrcv_disconnect()

static void libpqrcv_disconnect ( WalReceiverConn *  conn )

static

Definition at line 779 of file libpqwalreceiver.c.

{
    libpqsrv_disconnect(conn->streamConn);
    PQfreemem(conn->recvBuf);
    pfree(conn);
}

libpqrcv_endstreaming()

static void libpqrcv_endstreaming ( WalReceiverConn *  conn, TimeLineID *  next_tli  )

static

Definition at line 658 of file libpqwalreceiver.c.

{
    PGresult   *res;
 
    /*
     * Send copy-end message.  As in libpqsrv_exec, this could theoretically
     * block, but the risk seems small.
     */
    if (PQputCopyEnd(conn->streamConn, NULL) <= 0 ||
        PQflush(conn->streamConn))
        ereport(ERROR,
                (errcode(ERRCODE_CONNECTION_FAILURE),
                 errmsg("could not send end-of-streaming message to primary: %s",
                        pchomp(PQerrorMessage(conn->streamConn)))));
 
    *next_tli = 0;
 
    /*
     * After COPY is finished, we should receive a result set indicating the
     * next timeline's ID, or just CommandComplete if the server was shut
     * down.
     *
     * If we had not yet received CopyDone from the backend, PGRES_COPY_OUT is
     * also possible in case we aborted the copy in mid-stream.
     */
    res = libpqsrv_get_result(conn->streamConn,
                              WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
    if (PQresultStatus(res) == PGRES_TUPLES_OK)
    {
        /*
         * Read the next timeline's ID. The server also sends the timeline's
         * starting point, but it is ignored.
         */
        if (PQnfields(res) < 2 || PQntuples(res) != 1)
            ereport(ERROR,
                    (errcode(ERRCODE_PROTOCOL_VIOLATION),
                     errmsg("unexpected result set after end-of-streaming")));
        *next_tli = pg_strtoint32(PQgetvalue(res, 0, 0));
        PQclear(res);
 
        /* the result set should be followed by CommandComplete */
        res = libpqsrv_get_result(conn->streamConn,
                                  WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
    }
    else if (PQresultStatus(res) == PGRES_COPY_OUT)
    {
        PQclear(res);
 
        /* End the copy */
        if (PQendcopy(conn->streamConn))
            ereport(ERROR,
                    (errcode(ERRCODE_CONNECTION_FAILURE),
                     errmsg("error while shutting down streaming COPY: %s",
                            pchomp(PQerrorMessage(conn->streamConn)))));
 
        /* CommandComplete should follow */
        res = libpqsrv_get_result(conn->streamConn,
                                  WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
    }
 
    if (PQresultStatus(res) != PGRES_COMMAND_OK)
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg("error reading result of streaming command: %s",
                        pchomp(PQerrorMessage(conn->streamConn)))));
    PQclear(res);
 
    /* Verify that there are no more results */
    res = libpqsrv_get_result(conn->streamConn,
                              WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
    if (res != NULL)
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg("unexpected result after CommandComplete: %s",
                        pchomp(PQerrorMessage(conn->streamConn)))));
}

libpqrcv_exec()

static WalRcvExecResult * libpqrcv_exec ( WalReceiverConn *  conn, const char *  query, const int  nRetTypes, const Oid *  retTypes  )

static

Definition at line 1148 of file libpqwalreceiver.c.

{
    PGresult   *pgres = NULL;
    WalRcvExecResult *walres = palloc0_object(WalRcvExecResult);
    char       *diag_sqlstate;
 
    if (MyDatabaseId == InvalidOid)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("the query interface requires a database connection")));
 
    pgres = libpqsrv_exec(conn->streamConn,
                          query,
                          WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
 
    switch (PQresultStatus(pgres))
    {
        case PGRES_TUPLES_OK:
        case PGRES_SINGLE_TUPLE:
        case PGRES_TUPLES_CHUNK:
            walres->status = WALRCV_OK_TUPLES;
            libpqrcv_processTuples(pgres, walres, nRetTypes, retTypes);
            break;
 
        case PGRES_COPY_IN:
            walres->status = WALRCV_OK_COPY_IN;
            break;
 
        case PGRES_COPY_OUT:
            walres->status = WALRCV_OK_COPY_OUT;
            break;
 
        case PGRES_COPY_BOTH:
            walres->status = WALRCV_OK_COPY_BOTH;
            break;
 
        case PGRES_COMMAND_OK:
            walres->status = WALRCV_OK_COMMAND;
            break;
 
            /* Empty query is considered error. */
        case PGRES_EMPTY_QUERY:
            walres->status = WALRCV_ERROR;
            walres->err = _("empty query");
            break;
 
        case PGRES_PIPELINE_SYNC:
        case PGRES_PIPELINE_ABORTED:
            walres->status = WALRCV_ERROR;
            walres->err = _("unexpected pipeline mode");
            break;
 
        case PGRES_NONFATAL_ERROR:
        case PGRES_FATAL_ERROR:
        case PGRES_BAD_RESPONSE:
            walres->status = WALRCV_ERROR;
            walres->err = pchomp(PQerrorMessage(conn->streamConn));
            diag_sqlstate = PQresultErrorField(pgres, PG_DIAG_SQLSTATE);
            if (diag_sqlstate)
                walres->sqlstate = MAKE_SQLSTATE(diag_sqlstate[0],
                                                 diag_sqlstate[1],
                                                 diag_sqlstate[2],
                                                 diag_sqlstate[3],
                                                 diag_sqlstate[4]);
            break;
    }
 
    PQclear(pgres);
 
    return walres;
}

References _, conn, ereport, errcode(), errmsg, ERROR, fb(), InvalidOid, libpqrcv_processTuples(), libpqsrv_exec(), MAKE_SQLSTATE, MyDatabaseId, palloc0_object, pchomp(), PG_DIAG_SQLSTATE, PGRES_BAD_RESPONSE, PGRES_COMMAND_OK, PGRES_COPY_BOTH, PGRES_COPY_IN, PGRES_COPY_OUT, PGRES_EMPTY_QUERY, PGRES_FATAL_ERROR, PGRES_NONFATAL_ERROR, PGRES_PIPELINE_ABORTED, PGRES_PIPELINE_SYNC, PGRES_SINGLE_TUPLE, PGRES_TUPLES_CHUNK, PGRES_TUPLES_OK, PQclear, PQerrorMessage(), PQresultErrorField, PQresultStatus, WALRCV_ERROR, WALRCV_OK_COMMAND, WALRCV_OK_COPY_BOTH, WALRCV_OK_COPY_IN, WALRCV_OK_COPY_OUT, and WALRCV_OK_TUPLES.

libpqrcv_get_backend_pid()

static pid_t libpqrcv_get_backend_pid ( WalReceiverConn *  conn )

static

Definition at line 1062 of file libpqwalreceiver.c.

{
    return PQbackendPID(conn->streamConn);
}

References conn, and PQbackendPID().

libpqrcv_get_conninfo()

static char * libpqrcv_get_conninfo ( WalReceiverConn *  conn )

static

Definition at line 351 of file libpqwalreceiver.c.

{
    PQconninfoOption *conn_opts;
    PQconninfoOption *conn_opt;
    PQExpBufferData buf;
    char       *retval;
 
    Assert(conn->streamConn != NULL);
 
    initPQExpBuffer(&buf);
    conn_opts = PQconninfo(conn->streamConn);
 
    if (conn_opts == NULL)
        ereport(ERROR,
                (errcode(ERRCODE_OUT_OF_MEMORY),
                 errmsg("could not parse connection string: %s",
                        _("out of memory"))));
 
    /* build a clean connection string from pieces */
    for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++)
    {
        bool        obfuscate;
 
        /* Skip debug and empty options */
        if (strchr(conn_opt->dispchar, 'D') ||
            conn_opt->val == NULL ||
            conn_opt->val[0] == '\0')
            continue;
 
        /* Obfuscate security-sensitive options */
        obfuscate = strchr(conn_opt->dispchar, '*') != NULL;
 
        appendPQExpBuffer(&buf, "%s%s=%s",
                          buf.len == 0 ? "" : " ",
                          conn_opt->keyword,
                          obfuscate ? "********" : conn_opt->val);
    }
 
    PQconninfoFree(conn_opts);
 
    retval = PQExpBufferDataBroken(buf) ? NULL : pstrdup(buf.data);
    termPQExpBuffer(&buf);
    return retval;
}

References _, appendPQExpBuffer(), Assert, buf, conn, ereport, errcode(), errmsg, ERROR, fb(), initPQExpBuffer(), PQconninfo(), PQconninfoFree(), PQExpBufferDataBroken, pstrdup(), and termPQExpBuffer().

libpqrcv_get_dbname_from_conninfo()

static char * libpqrcv_get_dbname_from_conninfo ( const char *  connInfo )

static

Definition at line 475 of file libpqwalreceiver.c.

{
    return libpqrcv_get_option_from_conninfo(connInfo, "dbname");
}

References fb(), and libpqrcv_get_option_from_conninfo().

libpqrcv_get_option_from_conninfo()

static char * libpqrcv_get_option_from_conninfo ( const char *  connInfo, const char *  keyword  )

static

Definition at line 487 of file libpqwalreceiver.c.

{
    PQconninfoOption *opts;
    char       *option = NULL;
    char       *err = NULL;
 
    opts = PQconninfoParse(connInfo, &err);
    if (opts == NULL)
    {
        /* The error string is malloc'd, so we must free it explicitly */
        char       *errcopy = err ? pstrdup(err) : "out of memory";
 
        PQfreemem(err);
        ereport(ERROR,
                (errcode(ERRCODE_SYNTAX_ERROR),
                 errmsg("invalid connection string syntax: %s", errcopy)));
    }
 
    for (PQconninfoOption *opt = opts; opt->keyword != NULL; ++opt)
    {
        /*
         * If the same option appears multiple times, then the last one will
         * be returned
         */
        if (strcmp(opt->keyword, keyword) == 0 && opt->val &&
            *opt->val)
        {
            if (option)
                pfree(option);
 
            option = pstrdup(opt->val);
        }
    }
 
    PQconninfoFree(opts);
    return option;
}

References ereport, err(), errcode(), errmsg, ERROR, fb(), opts, pfree(), PQconninfoFree(), PQconninfoParse(), PQfreemem(), and pstrdup().

Referenced by libpqrcv_connect(), and libpqrcv_get_dbname_from_conninfo().

libpqrcv_get_senderinfo()

static void libpqrcv_get_senderinfo ( WalReceiverConn *  conn, char **  sender_host, int *  sender_port  )

static

Definition at line 400 of file libpqwalreceiver.c.

{
    char       *ret = NULL;
 
    *sender_host = NULL;
    *sender_port = 0;
 
    Assert(conn->streamConn != NULL);
 
    ret = PQhost(conn->streamConn);
    if (ret && strlen(ret) != 0)
        *sender_host = pstrdup(ret);
 
    ret = PQport(conn->streamConn);
    if (ret && strlen(ret) != 0)
        *sender_port = atoi(ret);
}

References Assert, conn, fb(), PQhost(), PQport(), and pstrdup().

libpqrcv_identify_system()

static char * libpqrcv_identify_system ( WalReceiverConn *  conn, TimeLineID *  primary_tli  )

static

Definition at line 424 of file libpqwalreceiver.c.

{
    PGresult   *res;
    char       *primary_sysid;
 
    /*
     * Get the system identifier and timeline ID as a DataRow message from the
     * primary server.
     */
    res = libpqsrv_exec(conn->streamConn,
                        "IDENTIFY_SYSTEM",
                        WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg("could not receive database system identifier and timeline ID from "
                        "the primary server: %s",
                        pchomp(PQerrorMessage(conn->streamConn)))));
 
    /*
     * IDENTIFY_SYSTEM returns 3 columns in 9.3 and earlier, and 4 columns in
     * 9.4 and onwards.
     */
    if (PQnfields(res) < 3 || PQntuples(res) != 1)
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg("invalid response from primary server"),
                 errdetail("Could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields.",
                           PQntuples(res), PQnfields(res), 1, 3)));
    primary_sysid = pstrdup(PQgetvalue(res, 0, 0));
    *primary_tli = pg_strtoint32(PQgetvalue(res, 0, 1));
    PQclear(res);
 
    return primary_sysid;
}

References conn, ereport, errcode(), ERRCODE_PROTOCOL_VIOLATION, errdetail(), errmsg, ERROR, fb(), libpqsrv_exec(), pchomp(), pg_strtoint32(), PGRES_TUPLES_OK, PQclear, PQerrorMessage(), PQgetvalue, PQnfields, PQntuples, PQresultStatus, and pstrdup().

libpqrcv_processTuples()

static void libpqrcv_processTuples ( PGresult *  pgres, WalRcvExecResult *  walres, const int  nRetTypes, const Oid *  retTypes  )

static

Definition at line 1071 of file libpqwalreceiver.c.

{
    int         tupn;
    int         coln;
    int         nfields = PQnfields(pgres);
    HeapTuple   tuple;
    AttInMetadata *attinmeta;
    MemoryContext rowcontext;
    MemoryContext oldcontext;
 
    /* Make sure we got expected number of fields. */
    if (nfields != nRetTypes)
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg("invalid query response"),
                 errdetail("Expected %d fields, got %d fields.",
                           nRetTypes, nfields)));
 
    walres->tuplestore = tuplestore_begin_heap(true, false, work_mem);
 
    /* Create tuple descriptor corresponding to expected result. */
    walres->tupledesc = CreateTemplateTupleDesc(nRetTypes);
    for (coln = 0; coln < nRetTypes; coln++)
        TupleDescInitEntry(walres->tupledesc, (AttrNumber) coln + 1,
                           PQfname(pgres, coln), retTypes[coln], -1, 0);
    TupleDescFinalize(walres->tupledesc);
    attinmeta = TupleDescGetAttInMetadata(walres->tupledesc);
 
    /* No point in doing more here if there were no tuples returned. */
    if (PQntuples(pgres) == 0)
        return;
 
    /* Create temporary context for local allocations. */
    rowcontext = AllocSetContextCreate(CurrentMemoryContext,
                                       "libpqrcv query result context",
                                       ALLOCSET_DEFAULT_SIZES);
 
    /* Process returned rows. */
    for (tupn = 0; tupn < PQntuples(pgres); tupn++)
    {
        char       *cstrs[MaxTupleAttributeNumber];
 
        CHECK_FOR_INTERRUPTS();
 
        /* Do the allocations in temporary context. */
        oldcontext = MemoryContextSwitchTo(rowcontext);
 
        /*
         * Fill cstrs with null-terminated strings of column values.
         */
        for (coln = 0; coln < nfields; coln++)
        {
            if (PQgetisnull(pgres, tupn, coln))
                cstrs[coln] = NULL;
            else
                cstrs[coln] = PQgetvalue(pgres, tupn, coln);
        }
 
        /* Convert row to a tuple, and add it to the tuplestore */
        tuple = BuildTupleFromCStrings(attinmeta, cstrs);
        tuplestore_puttuple(walres->tuplestore, tuple);
 
        /* Clean up */
        MemoryContextSwitchTo(oldcontext);
        MemoryContextReset(rowcontext);
    }
 
    MemoryContextDelete(rowcontext);
}

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, BuildTupleFromCStrings(), CHECK_FOR_INTERRUPTS, CreateTemplateTupleDesc(), CurrentMemoryContext, ereport, errcode(), ERRCODE_PROTOCOL_VIOLATION, errdetail(), errmsg, ERROR, fb(), MaxTupleAttributeNumber, MemoryContextDelete(), MemoryContextReset(), MemoryContextSwitchTo(), PQfname, PQgetisnull, PQgetvalue, PQnfields, PQntuples, TupleDescFinalize(), TupleDescGetAttInMetadata(), TupleDescInitEntry(), tuplestore_begin_heap(), tuplestore_puttuple(), and work_mem.

Referenced by libpqrcv_exec().

libpqrcv_readtimelinehistoryfile()

static void libpqrcv_readtimelinehistoryfile ( WalReceiverConn *  conn, TimeLineID  tli, char **  filename, char **  content, int *  len  )

static

Definition at line 739 of file libpqwalreceiver.c.

{
    PGresult   *res;
    char        cmd[64];
 
    Assert(!conn->logical);
 
    /*
     * Request the primary to send over the history file for given timeline.
     */
    snprintf(cmd, sizeof(cmd), "TIMELINE_HISTORY %u", tli);
    res = libpqsrv_exec(conn->streamConn,
                        cmd,
                        WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
    if (PQresultStatus(res) != PGRES_TUPLES_OK)
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg("could not receive timeline history file from "
                        "the primary server: %s",
                        pchomp(PQerrorMessage(conn->streamConn)))));
    if (PQnfields(res) != 2 || PQntuples(res) != 1)
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg("invalid response from primary server"),
                 errdetail("Expected 1 tuple with 2 fields, got %d tuples with %d fields.",
                           PQntuples(res), PQnfields(res))));
    *filename = pstrdup(PQgetvalue(res, 0, 0));
 
    *len = PQgetlength(res, 0, 1);
    *content = palloc(*len);
    memcpy(*content, PQgetvalue(res, 0, 1), *len);
    PQclear(res);
}

libpqrcv_receive()

static int libpqrcv_receive ( WalReceiverConn *  conn, char **  buffer, pgsocket *  wait_fd  )

static

Definition at line 803 of file libpqwalreceiver.c.

{
    int         rawlen;
 
    PQfreemem(conn->recvBuf);
    conn->recvBuf = NULL;
 
    /* Try to receive a CopyData message */
    rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
    if (rawlen == 0)
    {
        /* Try consuming some data. */
        if (PQconsumeInput(conn->streamConn) == 0)
            ereport(ERROR,
                    (errcode(ERRCODE_CONNECTION_FAILURE),
                     errmsg("could not receive data from WAL stream: %s",
                            pchomp(PQerrorMessage(conn->streamConn)))));
 
        /* Now that we've consumed some input, try again */
        rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
        if (rawlen == 0)
        {
            /* Tell caller to try again when our socket is ready. */
            *wait_fd = PQsocket(conn->streamConn);
            return 0;
        }
    }
    if (rawlen == -1)           /* end-of-streaming or error */
    {
        PGresult   *res;
 
        res = libpqsrv_get_result(conn->streamConn,
                                  WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
        if (PQresultStatus(res) == PGRES_COMMAND_OK)
        {
            PQclear(res);
 
            /* Verify that there are no more results. */
            res = libpqsrv_get_result(conn->streamConn,
                                      WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
            if (res != NULL)
            {
                PQclear(res);
 
                /*
                 * If the other side closed the connection orderly (otherwise
                 * we'd seen an error, or PGRES_COPY_IN) don't report an error
                 * here, but let callers deal with it.
                 */
                if (PQstatus(conn->streamConn) == CONNECTION_BAD)
                    return -1;
 
                ereport(ERROR,
                        (errcode(ERRCODE_PROTOCOL_VIOLATION),
                         errmsg("unexpected result after CommandComplete: %s",
                                PQerrorMessage(conn->streamConn))));
            }
 
            return -1;
        }
        else if (PQresultStatus(res) == PGRES_COPY_IN)
        {
            PQclear(res);
            return -1;
        }
        else
            ereport(ERROR,
                    (errcode(ERRCODE_PROTOCOL_VIOLATION),
                     errmsg("could not receive data from WAL stream: %s",
                            pchomp(PQerrorMessage(conn->streamConn)))));
    }
    if (rawlen < -1)
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg("could not receive data from WAL stream: %s",
                        pchomp(PQerrorMessage(conn->streamConn)))));
 
    /* Return received messages to caller */
    *buffer = conn->recvBuf;
    return rawlen;
}

libpqrcv_send()

static void libpqrcv_send ( WalReceiverConn *  conn, const char *  buffer, int  nbytes  )

static

Definition at line 892 of file libpqwalreceiver.c.

{
    if (PQputCopyData(conn->streamConn, buffer, nbytes) <= 0 ||
        PQflush(conn->streamConn))
        ereport(ERROR,
                (errcode(ERRCODE_CONNECTION_FAILURE),
                 errmsg("could not send data to WAL stream: %s",
                        pchomp(PQerrorMessage(conn->streamConn)))));
}

libpqrcv_server_version()

static int libpqrcv_server_version ( WalReceiverConn *  conn )

static

Definition at line 464 of file libpqwalreceiver.c.

{
    return PQserverVersion(conn->streamConn);
}

References conn, and PQserverVersion().

libpqrcv_startstreaming()

static bool libpqrcv_startstreaming ( WalReceiverConn *  conn, const WalRcvStreamOptions *  options  )

static

Definition at line 562 of file libpqwalreceiver.c.

{
    StringInfoData cmd;
    PGresult   *res;
 
    Assert(options->logical == conn->logical);
    Assert(options->slotname || !options->logical);
 
    initStringInfo(&cmd);
 
    /* Build the command. */
    appendStringInfoString(&cmd, "START_REPLICATION");
    if (options->slotname != NULL)
    {
        appendStringInfoString(&cmd, " SLOT ");
        appendQuotedIdentifier(&cmd, options->slotname);
    }
 
    if (options->logical)
        appendStringInfoString(&cmd, " LOGICAL");
 
    appendStringInfo(&cmd, " %X/%08X", LSN_FORMAT_ARGS(options->startpoint));
 
    /*
     * Additional options are different depending on if we are doing logical
     * or physical replication.
     */
    if (options->logical)
    {
        char       *pubnames_str;
        List       *pubnames;
 
        appendStringInfoString(&cmd, " (");
 
        appendStringInfo(&cmd, "proto_version '%u'",
                         options->proto.logical.proto_version);
 
        if (options->proto.logical.streaming_str)
        {
            appendStringInfoString(&cmd, ", streaming ");
            appendQuotedLiteral(&cmd, options->proto.logical.streaming_str);
        }
 
        if (options->proto.logical.twophase &&
            PQserverVersion(conn->streamConn) >= 150000)
            appendStringInfoString(&cmd, ", two_phase 'on'");
 
        if (options->proto.logical.origin &&
            PQserverVersion(conn->streamConn) >= 160000)
        {
            appendStringInfoString(&cmd, ", origin ");
            appendQuotedLiteral(&cmd, options->proto.logical.origin);
        }
 
        pubnames = options->proto.logical.publication_names;
        pubnames_str = stringlist_to_identifierstr(pubnames);
        appendStringInfoString(&cmd, ", publication_names ");
        appendQuotedLiteral(&cmd, pubnames_str);
        pfree(pubnames_str);
 
        if (options->proto.logical.binary &&
            PQserverVersion(conn->streamConn) >= 140000)
            appendStringInfoString(&cmd, ", binary 'true'");
 
        appendStringInfoChar(&cmd, ')');
    }
    else
        appendStringInfo(&cmd, " TIMELINE %u",
                         options->proto.physical.startpointTLI);
 
    /* Start streaming. */
    res = libpqsrv_exec(conn->streamConn,
                        cmd.data,
                        WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
    pfree(cmd.data);
 
    if (PQresultStatus(res) == PGRES_COMMAND_OK)
    {
        PQclear(res);
        return false;
    }
    else if (PQresultStatus(res) != PGRES_COPY_BOTH)
        ereport(ERROR,
                (errcode(ERRCODE_PROTOCOL_VIOLATION),
                 errmsg("could not start WAL streaming: %s",
                        pchomp(PQerrorMessage(conn->streamConn)))));
    PQclear(res);
    return true;
}

PG_MODULE_MAGIC_EXT()

PG_MODULE_MAGIC_EXT	(	.	name = "libpqwalreceiver",
		.	version = PG_VERSION
	)

stringlist_to_identifierstr()

static char * stringlist_to_identifierstr ( List *  strings )

static

Definition at line 1230 of file libpqwalreceiver.c.

{
    ListCell   *lc;
    StringInfoData res;
    bool        first = true;
 
    initStringInfo(&res);
 
    foreach(lc, strings)
    {
        char       *val = strVal(lfirst(lc));
 
        if (first)
            first = false;
        else
            appendStringInfoChar(&res, ',');
        appendQuotedIdentifier(&res, val);
    }
 
    return res.data;
}

References appendQuotedIdentifier, appendStringInfoChar(), StringInfoData::data, fb(), initStringInfo(), lfirst, strVal, and val.

Variable Documentation

PQWalReceiverFunctions

WalReceiverFunctionsType PQWalReceiverFunctions

static

Initial value:

= {
    .walrcv_connect = libpqrcv_connect,
    .walrcv_check_conninfo = libpqrcv_check_conninfo,
    .walrcv_get_conninfo = libpqrcv_get_conninfo,
    .walrcv_get_senderinfo = libpqrcv_get_senderinfo,
    .walrcv_identify_system = libpqrcv_identify_system,
    .walrcv_server_version = libpqrcv_server_version,
    .walrcv_readtimelinehistoryfile = libpqrcv_readtimelinehistoryfile,
    .walrcv_startstreaming = libpqrcv_startstreaming,
    .walrcv_endstreaming = libpqrcv_endstreaming,
    .walrcv_receive = libpqrcv_receive,
    .walrcv_send = libpqrcv_send,
    .walrcv_create_slot = libpqrcv_create_slot,
    .walrcv_alter_slot = libpqrcv_alter_slot,
    .walrcv_get_dbname_from_conninfo = libpqrcv_get_dbname_from_conninfo,
    .walrcv_get_backend_pid = libpqrcv_get_backend_pid,
    .walrcv_exec = libpqrcv_exec,
    .walrcv_disconnect = libpqrcv_disconnect
}

Definition at line 98 of file libpqwalreceiver.c.

                                                         {
    .walrcv_connect = libpqrcv_connect,
    .walrcv_check_conninfo = libpqrcv_check_conninfo,
    .walrcv_get_conninfo = libpqrcv_get_conninfo,
    .walrcv_get_senderinfo = libpqrcv_get_senderinfo,
    .walrcv_identify_system = libpqrcv_identify_system,
    .walrcv_server_version = libpqrcv_server_version,
    .walrcv_readtimelinehistoryfile = libpqrcv_readtimelinehistoryfile,
    .walrcv_startstreaming = libpqrcv_startstreaming,
    .walrcv_endstreaming = libpqrcv_endstreaming,
    .walrcv_receive = libpqrcv_receive,
    .walrcv_send = libpqrcv_send,
    .walrcv_create_slot = libpqrcv_create_slot,
    .walrcv_alter_slot = libpqrcv_alter_slot,
    .walrcv_get_dbname_from_conninfo = libpqrcv_get_dbname_from_conninfo,
    .walrcv_get_backend_pid = libpqrcv_get_backend_pid,
    .walrcv_exec = libpqrcv_exec,
    .walrcv_disconnect = libpqrcv_disconnect
};

Referenced by _PG_init().