PostgreSQL Source Code  git master
libpqwalreceiver.c File Reference
#include "postgres.h"
#include <unistd.h>
#include <sys/time.h>
#include "access/xlog.h"
#include "catalog/pg_type.h"
#include "common/connect.h"
#include "funcapi.h"
#include "libpq-fe.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "pqexpbuffer.h"
#include "replication/walreceiver.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
 

Functions

void _PG_init (void)
 
static WalReceiverConnlibpqrcv_connect (const char *conninfo, bool logical, const char *appname, char **err)
 
static void libpqrcv_check_conninfo (const char *conninfo)
 
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 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, CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)
 
static pid_t libpqrcv_get_backend_pid (WalReceiverConn *conn)
 
static WalRcvExecResultlibpqrcv_exec (WalReceiverConn *conn, const char *query, const int nRetTypes, const Oid *retTypes)
 
static void libpqrcv_disconnect (WalReceiverConn *conn)
 
static PGresultlibpqrcv_PQexec (PGconn *streamConn, const char *query)
 
static PGresultlibpqrcv_PQgetResult (PGconn *streamConn)
 
static char * stringlist_to_identifierstr (PGconn *conn, List *strings)
 
static void libpqrcv_processTuples (PGresult *pgres, WalRcvExecResult *walres, const int nRetTypes, const Oid *retTypes)
 

Variables

 PG_MODULE_MAGIC
 
static WalReceiverFunctionsType PQWalReceiverFunctions
 

Function Documentation

◆ _PG_init()

void _PG_init ( void  )

Definition at line 112 of file libpqwalreceiver.c.

References elog, ERROR, PQWalReceiverFunctions, and WalReceiverFunctions.

113 {
114  if (WalReceiverFunctions != NULL)
115  elog(ERROR, "libpqwalreceiver already loaded");
117 }
#define ERROR
Definition: elog.h:43
static WalReceiverFunctionsType PQWalReceiverFunctions
WalReceiverFunctionsType * WalReceiverFunctions
Definition: walreceiver.c:94
#define elog(elevel,...)
Definition: elog.h:214

◆ libpqrcv_check_conninfo()

static void libpqrcv_check_conninfo ( const char *  conninfo)
static

Definition at line 242 of file libpqwalreceiver.c.

References ereport, errcode(), errmsg(), ERROR, PQconninfoFree(), and PQconninfoParse().

243 {
244  PQconninfoOption *opts = NULL;
245  char *err = NULL;
246 
247  opts = PQconninfoParse(conninfo, &err);
248  if (opts == NULL)
249  ereport(ERROR,
250  (errcode(ERRCODE_SYNTAX_ERROR),
251  errmsg("invalid connection string syntax: %s", err)));
252 
253  PQconninfoFree(opts);
254 }
int errcode(int sqlerrcode)
Definition: elog.c:610
PQconninfoOption * PQconninfoParse(const char *conninfo, char **errmsg)
Definition: fe-connect.c:5274
#define ERROR
Definition: elog.h:43
void PQconninfoFree(PQconninfoOption *connOptions)
Definition: fe-connect.c:6504
#define ereport(elevel,...)
Definition: elog.h:144
int errmsg(const char *fmt,...)
Definition: elog.c:824

◆ libpqrcv_connect()

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

Definition at line 125 of file libpqwalreceiver.c.

References ALWAYS_SECURE_SEARCH_PATH_SQL, Assert, conn, CONNECTION_BAD, CONNECTION_OK, CONNECTION_STARTED, ereport, errmsg(), ERROR, GetDatabaseEncodingName(), i, libpqrcv_PQexec(), WalReceiverConn::logical, MyLatch, palloc0(), pchomp(), PGRES_POLLING_FAILED, PGRES_POLLING_OK, PGRES_POLLING_READING, PGRES_POLLING_WRITING, PGRES_TUPLES_OK, PQclear(), PQconnectPoll(), PQconnectStartParams(), PQerrorMessage(), PQresultStatus(), PQsocket(), PQstatus(), ProcessWalRcvInterrupts(), ResetLatch(), status(), WalReceiverConn::streamConn, WAIT_EVENT_LIBPQWALRECEIVER_CONNECT, WaitLatchOrSocket(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, WL_SOCKET_CONNECTED, WL_SOCKET_READABLE, and WL_SOCKET_WRITEABLE.

127 {
130  const char *keys[5];
131  const char *vals[5];
132  int i = 0;
133 
134  /*
135  * We use the expand_dbname parameter to process the connection string (or
136  * URI), and pass some extra options.
137  */
138  keys[i] = "dbname";
139  vals[i] = conninfo;
140  keys[++i] = "replication";
141  vals[i] = logical ? "database" : "true";
142  if (!logical)
143  {
144  /*
145  * The database name is ignored by the server in replication mode, but
146  * specify "replication" for .pgpass lookup.
147  */
148  keys[++i] = "dbname";
149  vals[i] = "replication";
150  }
151  keys[++i] = "fallback_application_name";
152  vals[i] = appname;
153  if (logical)
154  {
155  keys[++i] = "client_encoding";
156  vals[i] = GetDatabaseEncodingName();
157  }
158  keys[++i] = NULL;
159  vals[i] = NULL;
160 
161  Assert(i < sizeof(keys));
162 
163  conn = palloc0(sizeof(WalReceiverConn));
164  conn->streamConn = PQconnectStartParams(keys, vals,
165  /* expand_dbname = */ true);
166  if (PQstatus(conn->streamConn) == CONNECTION_BAD)
167  {
168  *err = pchomp(PQerrorMessage(conn->streamConn));
169  return NULL;
170  }
171 
172  /*
173  * Poll connection until we have OK or FAILED status.
174  *
175  * Per spec for PQconnectPoll, first wait till socket is write-ready.
176  */
177  status = PGRES_POLLING_WRITING;
178  do
179  {
180  int io_flag;
181  int rc;
182 
183  if (status == PGRES_POLLING_READING)
184  io_flag = WL_SOCKET_READABLE;
185 #ifdef WIN32
186  /* Windows needs a different test while waiting for connection-made */
187  else if (PQstatus(conn->streamConn) == CONNECTION_STARTED)
188  io_flag = WL_SOCKET_CONNECTED;
189 #endif
190  else
191  io_flag = WL_SOCKET_WRITEABLE;
192 
194  WL_EXIT_ON_PM_DEATH | WL_LATCH_SET | io_flag,
195  PQsocket(conn->streamConn),
196  0,
198 
199  /* Interrupted? */
200  if (rc & WL_LATCH_SET)
201  {
204  }
205 
206  /* If socket is ready, advance the libpq state machine */
207  if (rc & io_flag)
208  status = PQconnectPoll(conn->streamConn);
209  } while (status != PGRES_POLLING_OK && status != PGRES_POLLING_FAILED);
210 
211  if (PQstatus(conn->streamConn) != CONNECTION_OK)
212  {
213  *err = pchomp(PQerrorMessage(conn->streamConn));
214  return NULL;
215  }
216 
217  if (logical)
218  {
219  PGresult *res;
220 
221  res = libpqrcv_PQexec(conn->streamConn,
223  if (PQresultStatus(res) != PGRES_TUPLES_OK)
224  {
225  PQclear(res);
226  ereport(ERROR,
227  (errmsg("could not clear search path: %s",
228  pchomp(PQerrorMessage(conn->streamConn)))));
229  }
230  PQclear(res);
231  }
232 
233  conn->logical = logical;
234 
235  return conn;
236 }
#define WL_SOCKET_WRITEABLE
Definition: latch.h:126
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6672
void ProcessWalRcvInterrupts(void)
Definition: walreceiver.c:157
#define WL_SOCKET_READABLE
Definition: latch.h:125
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2692
void ResetLatch(Latch *latch)
Definition: latch.c:588
char * pchomp(const char *in)
Definition: mcxt.c:1214
#define ERROR
Definition: elog.h:43
PGconn * conn
Definition: streamutil.c:54
int WaitLatchOrSocket(Latch *latch, int wakeEvents, pgsocket sock, long timeout, uint32 wait_event_info)
Definition: latch.c:438
void * palloc0(Size size)
Definition: mcxt.c:980
PostgresPollingStatusType PQconnectPoll(PGconn *conn)
Definition: fe-connect.c:2259
static PGresult * libpqrcv_PQexec(PGconn *streamConn, const char *query)
#define ereport(elevel,...)
Definition: elog.h:144
void PQclear(PGresult *res)
Definition: fe-exec.c:694
#define Assert(condition)
Definition: c.h:745
const char * GetDatabaseEncodingName(void)
Definition: mbutils.c:1157
#define ALWAYS_SECURE_SEARCH_PATH_SQL
Definition: connect.h:25
PostgresPollingStatusType
Definition: libpq-fe.h:74
int errmsg(const char *fmt,...)
Definition: elog.c:824
int i
PGconn * PQconnectStartParams(const char *const *keywords, const char *const *values, int expand_dbname)
Definition: fe-connect.c:749
struct Latch * MyLatch
Definition: globals.c:54
static void static void status(const char *fmt,...) pg_attribute_printf(1
Definition: pg_regress.c:225
ConnStatusType PQstatus(const PGconn *conn)
Definition: fe-connect.c:6619
#define WL_SOCKET_CONNECTED
Definition: latch.h:134
#define WL_LATCH_SET
Definition: latch.h:124
int PQsocket(const PGconn *conn)
Definition: fe-connect.c:6690
#define WL_EXIT_ON_PM_DEATH
Definition: latch.h:129

◆ libpqrcv_create_slot()

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

Definition at line 827 of file libpqwalreceiver.c.

References appendStringInfo(), appendStringInfoString(), CRS_EXPORT_SNAPSHOT, CRS_NOEXPORT_SNAPSHOT, CRS_USE_SNAPSHOT, CStringGetDatum, StringInfoData::data, DatumGetLSN, DirectFunctionCall1Coll(), ereport, errmsg(), ERROR, initStringInfo(), InvalidOid, libpqrcv_PQexec(), WalReceiverConn::logical, pchomp(), pfree(), pg_lsn_in(), PGRES_TUPLES_OK, PQclear(), PQerrorMessage(), PQgetisnull(), PQgetvalue(), PQresultStatus(), pstrdup(), and WalReceiverConn::streamConn.

830 {
831  PGresult *res;
832  StringInfoData cmd;
833  char *snapshot;
834 
835  initStringInfo(&cmd);
836 
837  appendStringInfo(&cmd, "CREATE_REPLICATION_SLOT \"%s\"", slotname);
838 
839  if (temporary)
840  appendStringInfoString(&cmd, " TEMPORARY");
841 
842  if (conn->logical)
843  {
844  appendStringInfoString(&cmd, " LOGICAL pgoutput");
845  switch (snapshot_action)
846  {
847  case CRS_EXPORT_SNAPSHOT:
848  appendStringInfoString(&cmd, " EXPORT_SNAPSHOT");
849  break;
851  appendStringInfoString(&cmd, " NOEXPORT_SNAPSHOT");
852  break;
853  case CRS_USE_SNAPSHOT:
854  appendStringInfoString(&cmd, " USE_SNAPSHOT");
855  break;
856  }
857  }
858  else
859  {
860  appendStringInfoString(&cmd, " PHYSICAL RESERVE_WAL");
861  }
862 
863  res = libpqrcv_PQexec(conn->streamConn, cmd.data);
864  pfree(cmd.data);
865 
866  if (PQresultStatus(res) != PGRES_TUPLES_OK)
867  {
868  PQclear(res);
869  ereport(ERROR,
870  (errmsg("could not create replication slot \"%s\": %s",
871  slotname, pchomp(PQerrorMessage(conn->streamConn)))));
872  }
873 
874  if (lsn)
876  CStringGetDatum(PQgetvalue(res, 0, 1))));
877 
878  if (!PQgetisnull(res, 0, 2))
879  snapshot = pstrdup(PQgetvalue(res, 0, 2));
880  else
881  snapshot = NULL;
882 
883  PQclear(res);
884 
885  return snapshot;
886 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6672
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3163
char * pstrdup(const char *in)
Definition: mcxt.c:1186
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2692
Datum pg_lsn_in(PG_FUNCTION_ARGS)
Definition: pg_lsn.c:64
char * pchomp(const char *in)
Definition: mcxt.c:1214
void pfree(void *pointer)
Definition: mcxt.c:1056
void appendStringInfo(StringInfo str, const char *fmt,...)
Definition: stringinfo.c:91
#define ERROR
Definition: elog.h:43
void appendStringInfoString(StringInfo str, const char *s)
Definition: stringinfo.c:176
#define CStringGetDatum(X)
Definition: postgres.h:578
Datum DirectFunctionCall1Coll(PGFunction func, Oid collation, Datum arg1)
Definition: fmgr.c:795
void initStringInfo(StringInfo str)
Definition: stringinfo.c:59
static PGresult * libpqrcv_PQexec(PGconn *streamConn, const char *query)
#define InvalidOid
Definition: postgres_ext.h:36
#define DatumGetLSN(X)
Definition: pg_lsn.h:21
#define ereport(elevel,...)
Definition: elog.h:144
void PQclear(PGresult *res)
Definition: fe-exec.c:694
int errmsg(const char *fmt,...)
Definition: elog.c:824
int PQgetisnull(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3188

◆ libpqrcv_disconnect()

static void libpqrcv_disconnect ( WalReceiverConn conn)
static

Definition at line 700 of file libpqwalreceiver.c.

References pfree(), PQfinish(), PQfreemem(), WalReceiverConn::recvBuf, and WalReceiverConn::streamConn.

701 {
702  PQfinish(conn->streamConn);
703  if (conn->recvBuf != NULL)
704  PQfreemem(conn->recvBuf);
705  pfree(conn);
706 }
void PQfinish(PGconn *conn)
Definition: fe-connect.c:4170
void pfree(void *pointer)
Definition: mcxt.c:1056
void PQfreemem(void *ptr)
Definition: fe-exec.c:3296

◆ libpqrcv_endstreaming()

static void libpqrcv_endstreaming ( WalReceiverConn conn,
TimeLineID next_tli 
)
static

Definition at line 479 of file libpqwalreceiver.c.

References ereport, errmsg(), ERROR, libpqrcv_PQgetResult(), pchomp(), pg_strtoint32(), PGRES_COMMAND_OK, PGRES_COPY_OUT, PGRES_TUPLES_OK, PQclear(), PQendcopy(), PQerrorMessage(), PQflush(), PQgetvalue(), PQnfields(), PQntuples(), PQputCopyEnd(), PQresultStatus(), and WalReceiverConn::streamConn.

480 {
481  PGresult *res;
482 
483  /*
484  * Send copy-end message. As in libpqrcv_PQexec, this could theoretically
485  * block, but the risk seems small.
486  */
487  if (PQputCopyEnd(conn->streamConn, NULL) <= 0 ||
488  PQflush(conn->streamConn))
489  ereport(ERROR,
490  (errmsg("could not send end-of-streaming message to primary: %s",
491  pchomp(PQerrorMessage(conn->streamConn)))));
492 
493  *next_tli = 0;
494 
495  /*
496  * After COPY is finished, we should receive a result set indicating the
497  * next timeline's ID, or just CommandComplete if the server was shut
498  * down.
499  *
500  * If we had not yet received CopyDone from the backend, PGRES_COPY_OUT is
501  * also possible in case we aborted the copy in mid-stream.
502  */
503  res = libpqrcv_PQgetResult(conn->streamConn);
504  if (PQresultStatus(res) == PGRES_TUPLES_OK)
505  {
506  /*
507  * Read the next timeline's ID. The server also sends the timeline's
508  * starting point, but it is ignored.
509  */
510  if (PQnfields(res) < 2 || PQntuples(res) != 1)
511  ereport(ERROR,
512  (errmsg("unexpected result set after end-of-streaming")));
513  *next_tli = pg_strtoint32(PQgetvalue(res, 0, 0));
514  PQclear(res);
515 
516  /* the result set should be followed by CommandComplete */
517  res = libpqrcv_PQgetResult(conn->streamConn);
518  }
519  else if (PQresultStatus(res) == PGRES_COPY_OUT)
520  {
521  PQclear(res);
522 
523  /* End the copy */
524  if (PQendcopy(conn->streamConn))
525  ereport(ERROR,
526  (errmsg("error while shutting down streaming COPY: %s",
527  pchomp(PQerrorMessage(conn->streamConn)))));
528 
529  /* CommandComplete should follow */
530  res = libpqrcv_PQgetResult(conn->streamConn);
531  }
532 
533  if (PQresultStatus(res) != PGRES_COMMAND_OK)
534  ereport(ERROR,
535  (errmsg("error reading result of streaming command: %s",
536  pchomp(PQerrorMessage(conn->streamConn)))));
537  PQclear(res);
538 
539  /* Verify that there are no more results */
540  res = libpqrcv_PQgetResult(conn->streamConn);
541  if (res != NULL)
542  ereport(ERROR,
543  (errmsg("unexpected result after CommandComplete: %s",
544  pchomp(PQerrorMessage(conn->streamConn)))));
545 }
int PQnfields(const PGresult *res)
Definition: fe-exec.c:2777
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6672
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3163
int PQputCopyEnd(PGconn *conn, const char *errormsg)
Definition: fe-exec.c:2384
int PQntuples(const PGresult *res)
Definition: fe-exec.c:2769
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2692
static PGresult * libpqrcv_PQgetResult(PGconn *streamConn)
char * pchomp(const char *in)
Definition: mcxt.c:1214
#define ERROR
Definition: elog.h:43
int PQflush(PGconn *conn)
Definition: fe-exec.c:3283
#define ereport(elevel,...)
Definition: elog.h:144
void PQclear(PGresult *res)
Definition: fe-exec.c:694
int PQendcopy(PGconn *conn)
Definition: fe-exec.c:2618
int32 pg_strtoint32(const char *s)
Definition: numutils.c:263
int errmsg(const char *fmt,...)
Definition: elog.c:824

◆ libpqrcv_exec()

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

Definition at line 976 of file libpqwalreceiver.c.

References _, ereport, WalRcvExecResult::err, errcode(), errmsg(), ERROR, InvalidOid, libpqrcv_PQexec(), libpqrcv_processTuples(), MyDatabaseId, palloc0(), pchomp(), 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_SINGLE_TUPLE, PGRES_TUPLES_OK, PQclear(), PQerrorMessage(), PQresultStatus(), WalRcvExecResult::status, WalReceiverConn::streamConn, WALRCV_ERROR, WALRCV_OK_COMMAND, WALRCV_OK_COPY_BOTH, WALRCV_OK_COPY_IN, WALRCV_OK_COPY_OUT, and WALRCV_OK_TUPLES.

978 {
979  PGresult *pgres = NULL;
980  WalRcvExecResult *walres = palloc0(sizeof(WalRcvExecResult));
981 
982  if (MyDatabaseId == InvalidOid)
983  ereport(ERROR,
984  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
985  errmsg("the query interface requires a database connection")));
986 
987  pgres = libpqrcv_PQexec(conn->streamConn, query);
988 
989  switch (PQresultStatus(pgres))
990  {
991  case PGRES_SINGLE_TUPLE:
992  case PGRES_TUPLES_OK:
993  walres->status = WALRCV_OK_TUPLES;
994  libpqrcv_processTuples(pgres, walres, nRetTypes, retTypes);
995  break;
996 
997  case PGRES_COPY_IN:
998  walres->status = WALRCV_OK_COPY_IN;
999  break;
1000 
1001  case PGRES_COPY_OUT:
1002  walres->status = WALRCV_OK_COPY_OUT;
1003  break;
1004 
1005  case PGRES_COPY_BOTH:
1006  walres->status = WALRCV_OK_COPY_BOTH;
1007  break;
1008 
1009  case PGRES_COMMAND_OK:
1010  walres->status = WALRCV_OK_COMMAND;
1011  break;
1012 
1013  /* Empty query is considered error. */
1014  case PGRES_EMPTY_QUERY:
1015  walres->status = WALRCV_ERROR;
1016  walres->err = _("empty query");
1017  break;
1018 
1019  case PGRES_NONFATAL_ERROR:
1020  case PGRES_FATAL_ERROR:
1021  case PGRES_BAD_RESPONSE:
1022  walres->status = WALRCV_ERROR;
1023  walres->err = pchomp(PQerrorMessage(conn->streamConn));
1024  break;
1025  }
1026 
1027  PQclear(pgres);
1028 
1029  return walres;
1030 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6672
int errcode(int sqlerrcode)
Definition: elog.c:610
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2692
char * pchomp(const char *in)
Definition: mcxt.c:1214
#define ERROR
Definition: elog.h:43
static void libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres, const int nRetTypes, const Oid *retTypes)
void * palloc0(Size size)
Definition: mcxt.c:980
Oid MyDatabaseId
Definition: globals.c:85
static PGresult * libpqrcv_PQexec(PGconn *streamConn, const char *query)
#define InvalidOid
Definition: postgres_ext.h:36
#define ereport(elevel,...)
Definition: elog.h:144
void PQclear(PGresult *res)
Definition: fe-exec.c:694
WalRcvExecStatus status
Definition: walreceiver.h:211
int errmsg(const char *fmt,...)
Definition: elog.c:824
#define _(x)
Definition: elog.c:88

◆ libpqrcv_get_backend_pid()

static pid_t libpqrcv_get_backend_pid ( WalReceiverConn conn)
static

Definition at line 892 of file libpqwalreceiver.c.

References PQbackendPID(), and WalReceiverConn::streamConn.

893 {
894  return PQbackendPID(conn->streamConn);
895 }
int PQbackendPID(const PGconn *conn)
Definition: fe-connect.c:6698

◆ libpqrcv_get_conninfo()

static char * libpqrcv_get_conninfo ( WalReceiverConn conn)
static

Definition at line 261 of file libpqwalreceiver.c.

References _, appendPQExpBuffer(), Assert, buf, PQExpBufferData::data, _PQconninfoOption::dispchar, ereport, errmsg(), ERROR, initPQExpBuffer(), _PQconninfoOption::keyword, PQExpBufferData::len, PQconninfo(), PQconninfoFree(), PQExpBufferDataBroken, pstrdup(), WalReceiverConn::streamConn, termPQExpBuffer(), and _PQconninfoOption::val.

262 {
263  PQconninfoOption *conn_opts;
264  PQconninfoOption *conn_opt;
266  char *retval;
267 
268  Assert(conn->streamConn != NULL);
269 
270  initPQExpBuffer(&buf);
271  conn_opts = PQconninfo(conn->streamConn);
272 
273  if (conn_opts == NULL)
274  ereport(ERROR,
275  (errmsg("could not parse connection string: %s",
276  _("out of memory"))));
277 
278  /* build a clean connection string from pieces */
279  for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++)
280  {
281  bool obfuscate;
282 
283  /* Skip debug and empty options */
284  if (strchr(conn_opt->dispchar, 'D') ||
285  conn_opt->val == NULL ||
286  conn_opt->val[0] == '\0')
287  continue;
288 
289  /* Obfuscate security-sensitive options */
290  obfuscate = strchr(conn_opt->dispchar, '*') != NULL;
291 
292  appendPQExpBuffer(&buf, "%s%s=%s",
293  buf.len == 0 ? "" : " ",
294  conn_opt->keyword,
295  obfuscate ? "********" : conn_opt->val);
296  }
297 
298  PQconninfoFree(conn_opts);
299 
300  retval = PQExpBufferDataBroken(buf) ? NULL : pstrdup(buf.data);
301  termPQExpBuffer(&buf);
302  return retval;
303 }
void termPQExpBuffer(PQExpBuffer str)
Definition: pqexpbuffer.c:131
char * pstrdup(const char *in)
Definition: mcxt.c:1186
#define ERROR
Definition: elog.h:43
void appendPQExpBuffer(PQExpBuffer str, const char *fmt,...)
Definition: pqexpbuffer.c:267
static char * buf
Definition: pg_test_fsync.c:67
void PQconninfoFree(PQconninfoOption *connOptions)
Definition: fe-connect.c:6504
PQconninfoOption * PQconninfo(PGconn *conn)
Definition: fe-connect.c:6463
#define ereport(elevel,...)
Definition: elog.h:144
#define PQExpBufferDataBroken(buf)
Definition: pqexpbuffer.h:67
#define Assert(condition)
Definition: c.h:745
int errmsg(const char *fmt,...)
Definition: elog.c:824
#define _(x)
Definition: elog.c:88
void initPQExpBuffer(PQExpBuffer str)
Definition: pqexpbuffer.c:92

◆ libpqrcv_get_senderinfo()

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

Definition at line 309 of file libpqwalreceiver.c.

References Assert, PQhost(), PQport(), pstrdup(), and WalReceiverConn::streamConn.

311 {
312  char *ret = NULL;
313 
314  *sender_host = NULL;
315  *sender_port = 0;
316 
317  Assert(conn->streamConn != NULL);
318 
319  ret = PQhost(conn->streamConn);
320  if (ret && strlen(ret) != 0)
321  *sender_host = pstrdup(ret);
322 
323  ret = PQport(conn->streamConn);
324  if (ret && strlen(ret) != 0)
325  *sender_port = atoi(ret);
326 }
char * pstrdup(const char *in)
Definition: mcxt.c:1186
char * PQport(const PGconn *conn)
Definition: fe-connect.c:6591
char * PQhost(const PGconn *conn)
Definition: fe-connect.c:6555
#define Assert(condition)
Definition: c.h:745

◆ libpqrcv_identify_system()

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

Definition at line 333 of file libpqwalreceiver.c.

References ereport, errdetail(), errmsg(), ERROR, libpqrcv_PQexec(), pchomp(), pg_strtoint32(), PGRES_TUPLES_OK, PQclear(), PQerrorMessage(), PQgetvalue(), PQnfields(), PQntuples(), PQresultStatus(), pstrdup(), and WalReceiverConn::streamConn.

334 {
335  PGresult *res;
336  char *primary_sysid;
337 
338  /*
339  * Get the system identifier and timeline ID as a DataRow message from the
340  * primary server.
341  */
342  res = libpqrcv_PQexec(conn->streamConn, "IDENTIFY_SYSTEM");
343  if (PQresultStatus(res) != PGRES_TUPLES_OK)
344  {
345  PQclear(res);
346  ereport(ERROR,
347  (errmsg("could not receive database system identifier and timeline ID from "
348  "the primary server: %s",
349  pchomp(PQerrorMessage(conn->streamConn)))));
350  }
351  if (PQnfields(res) < 3 || PQntuples(res) != 1)
352  {
353  int ntuples = PQntuples(res);
354  int nfields = PQnfields(res);
355 
356  PQclear(res);
357  ereport(ERROR,
358  (errmsg("invalid response from primary server"),
359  errdetail("Could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields.",
360  ntuples, nfields, 3, 1)));
361  }
362  primary_sysid = pstrdup(PQgetvalue(res, 0, 0));
363  *primary_tli = pg_strtoint32(PQgetvalue(res, 0, 1));
364  PQclear(res);
365 
366  return primary_sysid;
367 }
int PQnfields(const PGresult *res)
Definition: fe-exec.c:2777
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6672
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3163
char * pstrdup(const char *in)
Definition: mcxt.c:1186
int PQntuples(const PGresult *res)
Definition: fe-exec.c:2769
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2692
char * pchomp(const char *in)
Definition: mcxt.c:1214
#define ERROR
Definition: elog.h:43
int errdetail(const char *fmt,...)
Definition: elog.c:957
static PGresult * libpqrcv_PQexec(PGconn *streamConn, const char *query)
#define ereport(elevel,...)
Definition: elog.h:144
void PQclear(PGresult *res)
Definition: fe-exec.c:694
int32 pg_strtoint32(const char *s)
Definition: numutils.c:263
int errmsg(const char *fmt,...)
Definition: elog.c:824

◆ libpqrcv_PQexec()

static PGresult * libpqrcv_PQexec ( PGconn streamConn,
const char *  query 
)
static

Definition at line 606 of file libpqwalreceiver.c.

References CONNECTION_BAD, libpqrcv_PQgetResult(), PGRES_COPY_BOTH, PGRES_COPY_IN, PGRES_COPY_OUT, PQclear(), PQresultStatus(), PQsendQuery(), and PQstatus().

Referenced by libpqrcv_connect(), libpqrcv_create_slot(), libpqrcv_exec(), libpqrcv_identify_system(), libpqrcv_readtimelinehistoryfile(), and libpqrcv_startstreaming().

607 {
608  PGresult *lastResult = NULL;
609 
610  /*
611  * PQexec() silently discards any prior query results on the connection.
612  * This is not required for this function as it's expected that the caller
613  * (which is this library in all cases) will behave correctly and we don't
614  * have to be backwards compatible with old libpq.
615  */
616 
617  /*
618  * Submit the query. Since we don't use non-blocking mode, this could
619  * theoretically block. In practice, since we don't send very long query
620  * strings, the risk seems negligible.
621  */
622  if (!PQsendQuery(streamConn, query))
623  return NULL;
624 
625  for (;;)
626  {
627  /* Wait for, and collect, the next PGresult. */
628  PGresult *result;
629 
630  result = libpqrcv_PQgetResult(streamConn);
631  if (result == NULL)
632  break; /* query is complete, or failure */
633 
634  /*
635  * Emulate PQexec()'s behavior of returning the last result when there
636  * are many. We are fine with returning just last error message.
637  */
638  PQclear(lastResult);
639  lastResult = result;
640 
641  if (PQresultStatus(lastResult) == PGRES_COPY_IN ||
642  PQresultStatus(lastResult) == PGRES_COPY_OUT ||
643  PQresultStatus(lastResult) == PGRES_COPY_BOTH ||
644  PQstatus(streamConn) == CONNECTION_BAD)
645  break;
646  }
647 
648  return lastResult;
649 }
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2692
static PGresult * libpqrcv_PQgetResult(PGconn *streamConn)
int PQsendQuery(PGconn *conn, const char *query)
Definition: fe-exec.c:1234
void PQclear(PGresult *res)
Definition: fe-exec.c:694
ConnStatusType PQstatus(const PGconn *conn)
Definition: fe-connect.c:6619

◆ libpqrcv_PQgetResult()

static PGresult * libpqrcv_PQgetResult ( PGconn streamConn)
static

Definition at line 655 of file libpqwalreceiver.c.

References MyLatch, PQconsumeInput(), PQgetResult(), PQisBusy(), PQsocket(), ProcessWalRcvInterrupts(), ResetLatch(), WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE, WaitLatchOrSocket(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and WL_SOCKET_READABLE.

Referenced by libpqrcv_endstreaming(), libpqrcv_PQexec(), and libpqrcv_receive().

656 {
657  /*
658  * Collect data until PQgetResult is ready to get the result without
659  * blocking.
660  */
661  while (PQisBusy(streamConn))
662  {
663  int rc;
664 
665  /*
666  * We don't need to break down the sleep into smaller increments,
667  * since we'll get interrupted by signals and can handle any
668  * interrupts here.
669  */
672  WL_LATCH_SET,
673  PQsocket(streamConn),
674  0,
676 
677  /* Interrupted? */
678  if (rc & WL_LATCH_SET)
679  {
682  }
683 
684  /* Consume whatever data is available from the socket */
685  if (PQconsumeInput(streamConn) == 0)
686  {
687  /* trouble; return NULL */
688  return NULL;
689  }
690  }
691 
692  /* Now we can collect and return the next PGresult */
693  return PQgetResult(streamConn);
694 }
void ProcessWalRcvInterrupts(void)
Definition: walreceiver.c:157
#define WL_SOCKET_READABLE
Definition: latch.h:125
void ResetLatch(Latch *latch)
Definition: latch.c:588
int WaitLatchOrSocket(Latch *latch, int wakeEvents, pgsocket sock, long timeout, uint32 wait_event_info)
Definition: latch.c:438
int PQconsumeInput(PGconn *conn)
Definition: fe-exec.c:1704
int PQisBusy(PGconn *conn)
Definition: fe-exec.c:1754
struct Latch * MyLatch
Definition: globals.c:54
#define WL_LATCH_SET
Definition: latch.h:124
int PQsocket(const PGconn *conn)
Definition: fe-connect.c:6690
PGresult * PQgetResult(PGconn *conn)
Definition: fe-exec.c:1778
#define WL_EXIT_ON_PM_DEATH
Definition: latch.h:129

◆ libpqrcv_processTuples()

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

Definition at line 901 of file libpqwalreceiver.c.

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, BuildTupleFromCStrings(), CreateTemplateTupleDesc(), CurrentMemoryContext, ereport, errdetail(), errmsg(), ERROR, MaxTupleAttributeNumber, MemoryContextDelete(), MemoryContextReset(), MemoryContextSwitchTo(), PQfname(), PQgetisnull(), PQgetvalue(), PQnfields(), PQntuples(), ProcessWalRcvInterrupts(), WalRcvExecResult::tupledesc, TupleDescGetAttInMetadata(), TupleDescInitEntry(), WalRcvExecResult::tuplestore, tuplestore_begin_heap(), tuplestore_puttuple(), and work_mem.

Referenced by libpqrcv_exec().

903 {
904  int tupn;
905  int coln;
906  int nfields = PQnfields(pgres);
907  HeapTuple tuple;
908  AttInMetadata *attinmeta;
909  MemoryContext rowcontext;
910  MemoryContext oldcontext;
911 
912  /* Make sure we got expected number of fields. */
913  if (nfields != nRetTypes)
914  ereport(ERROR,
915  (errmsg("invalid query response"),
916  errdetail("Expected %d fields, got %d fields.",
917  nRetTypes, nfields)));
918 
919  walres->tuplestore = tuplestore_begin_heap(true, false, work_mem);
920 
921  /* Create tuple descriptor corresponding to expected result. */
922  walres->tupledesc = CreateTemplateTupleDesc(nRetTypes);
923  for (coln = 0; coln < nRetTypes; coln++)
924  TupleDescInitEntry(walres->tupledesc, (AttrNumber) coln + 1,
925  PQfname(pgres, coln), retTypes[coln], -1, 0);
926  attinmeta = TupleDescGetAttInMetadata(walres->tupledesc);
927 
928  /* No point in doing more here if there were no tuples returned. */
929  if (PQntuples(pgres) == 0)
930  return;
931 
932  /* Create temporary context for local allocations. */
934  "libpqrcv query result context",
936 
937  /* Process returned rows. */
938  for (tupn = 0; tupn < PQntuples(pgres); tupn++)
939  {
940  char *cstrs[MaxTupleAttributeNumber];
941 
943 
944  /* Do the allocations in temporary context. */
945  oldcontext = MemoryContextSwitchTo(rowcontext);
946 
947  /*
948  * Fill cstrs with null-terminated strings of column values.
949  */
950  for (coln = 0; coln < nfields; coln++)
951  {
952  if (PQgetisnull(pgres, tupn, coln))
953  cstrs[coln] = NULL;
954  else
955  cstrs[coln] = PQgetvalue(pgres, tupn, coln);
956  }
957 
958  /* Convert row to a tuple, and add it to the tuplestore */
959  tuple = BuildTupleFromCStrings(attinmeta, cstrs);
960  tuplestore_puttuple(walres->tuplestore, tuple);
961 
962  /* Clean up */
963  MemoryContextSwitchTo(oldcontext);
964  MemoryContextReset(rowcontext);
965  }
966 
967  MemoryContextDelete(rowcontext);
968 }
int PQnfields(const PGresult *res)
Definition: fe-exec.c:2777
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:211
#define AllocSetContextCreate
Definition: memutils.h:170
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3163
#define MaxTupleAttributeNumber
Definition: htup_details.h:33
TupleDesc CreateTemplateTupleDesc(int natts)
Definition: tupdesc.c:44
char * PQfname(const PGresult *res, int field_num)
Definition: fe-exec.c:2855
void ProcessWalRcvInterrupts(void)
Definition: walreceiver.c:157
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
void MemoryContextReset(MemoryContext context)
Definition: mcxt.c:136
int PQntuples(const PGresult *res)
Definition: fe-exec.c:2769
HeapTuple BuildTupleFromCStrings(AttInMetadata *attinmeta, char **values)
Definition: execTuples.c:2116
TupleDesc tupledesc
Definition: walreceiver.h:214
#define ERROR
Definition: elog.h:43
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:192
void tuplestore_puttuple(Tuplestorestate *state, HeapTuple tuple)
Definition: tuplestore.c:730
int errdetail(const char *fmt,...)
Definition: elog.c:957
MemoryContext CurrentMemoryContext
Definition: mcxt.c:38
void TupleDescInitEntry(TupleDesc desc, AttrNumber attributeNumber, const char *attributeName, Oid oidtypeid, int32 typmod, int attdim)
Definition: tupdesc.c:603
Tuplestorestate * tuplestore_begin_heap(bool randomAccess, bool interXact, int maxKBytes)
Definition: tuplestore.c:318
AttInMetadata * TupleDescGetAttInMetadata(TupleDesc tupdesc)
Definition: execTuples.c:2067
int work_mem
Definition: globals.c:121
#define ereport(elevel,...)
Definition: elog.h:144
Tuplestorestate * tuplestore
Definition: walreceiver.h:213
int errmsg(const char *fmt,...)
Definition: elog.c:824
int PQgetisnull(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3188
int16 AttrNumber
Definition: attnum.h:21

◆ libpqrcv_readtimelinehistoryfile()

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

Definition at line 551 of file libpqwalreceiver.c.

References Assert, ereport, errdetail(), errmsg(), ERROR, libpqrcv_PQexec(), WalReceiverConn::logical, palloc(), pchomp(), PGRES_TUPLES_OK, PQclear(), PQerrorMessage(), PQgetlength(), PQgetvalue(), PQnfields(), PQntuples(), PQresultStatus(), pstrdup(), snprintf, and WalReceiverConn::streamConn.

554 {
555  PGresult *res;
556  char cmd[64];
557 
558  Assert(!conn->logical);
559 
560  /*
561  * Request the primary to send over the history file for given timeline.
562  */
563  snprintf(cmd, sizeof(cmd), "TIMELINE_HISTORY %u", tli);
564  res = libpqrcv_PQexec(conn->streamConn, cmd);
565  if (PQresultStatus(res) != PGRES_TUPLES_OK)
566  {
567  PQclear(res);
568  ereport(ERROR,
569  (errmsg("could not receive timeline history file from "
570  "the primary server: %s",
571  pchomp(PQerrorMessage(conn->streamConn)))));
572  }
573  if (PQnfields(res) != 2 || PQntuples(res) != 1)
574  {
575  int ntuples = PQntuples(res);
576  int nfields = PQnfields(res);
577 
578  PQclear(res);
579  ereport(ERROR,
580  (errmsg("invalid response from primary server"),
581  errdetail("Expected 1 tuple with 2 fields, got %d tuples with %d fields.",
582  ntuples, nfields)));
583  }
584  *filename = pstrdup(PQgetvalue(res, 0, 0));
585 
586  *len = PQgetlength(res, 0, 1);
587  *content = palloc(*len);
588  memcpy(*content, PQgetvalue(res, 0, 1), *len);
589  PQclear(res);
590 }
int PQgetlength(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3174
int PQnfields(const PGresult *res)
Definition: fe-exec.c:2777
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6672
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3163
char * pstrdup(const char *in)
Definition: mcxt.c:1186
int PQntuples(const PGresult *res)
Definition: fe-exec.c:2769
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2692
char * pchomp(const char *in)
Definition: mcxt.c:1214
#define ERROR
Definition: elog.h:43
int errdetail(const char *fmt,...)
Definition: elog.c:957
static PGresult * libpqrcv_PQexec(PGconn *streamConn, const char *query)
#define ereport(elevel,...)
Definition: elog.h:144
void PQclear(PGresult *res)
Definition: fe-exec.c:694
#define Assert(condition)
Definition: c.h:745
static char * filename
Definition: pg_dumpall.c:90
void * palloc(Size size)
Definition: mcxt.c:949
int errmsg(const char *fmt,...)
Definition: elog.c:824
#define snprintf
Definition: port.h:193

◆ libpqrcv_receive()

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

Definition at line 725 of file libpqwalreceiver.c.

References CONNECTION_BAD, ereport, errmsg(), ERROR, libpqrcv_PQgetResult(), pchomp(), PGRES_COMMAND_OK, PGRES_COPY_IN, PQclear(), PQconsumeInput(), PQerrorMessage(), PQfreemem(), PQgetCopyData(), PQresultStatus(), PQsocket(), PQstatus(), WalReceiverConn::recvBuf, and WalReceiverConn::streamConn.

727 {
728  int rawlen;
729 
730  if (conn->recvBuf != NULL)
731  PQfreemem(conn->recvBuf);
732  conn->recvBuf = NULL;
733 
734  /* Try to receive a CopyData message */
735  rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
736  if (rawlen == 0)
737  {
738  /* Try consuming some data. */
739  if (PQconsumeInput(conn->streamConn) == 0)
740  ereport(ERROR,
741  (errmsg("could not receive data from WAL stream: %s",
742  pchomp(PQerrorMessage(conn->streamConn)))));
743 
744  /* Now that we've consumed some input, try again */
745  rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
746  if (rawlen == 0)
747  {
748  /* Tell caller to try again when our socket is ready. */
749  *wait_fd = PQsocket(conn->streamConn);
750  return 0;
751  }
752  }
753  if (rawlen == -1) /* end-of-streaming or error */
754  {
755  PGresult *res;
756 
757  res = libpqrcv_PQgetResult(conn->streamConn);
758  if (PQresultStatus(res) == PGRES_COMMAND_OK)
759  {
760  PQclear(res);
761 
762  /* Verify that there are no more results. */
763  res = libpqrcv_PQgetResult(conn->streamConn);
764  if (res != NULL)
765  {
766  PQclear(res);
767 
768  /*
769  * If the other side closed the connection orderly (otherwise
770  * we'd seen an error, or PGRES_COPY_IN) don't report an error
771  * here, but let callers deal with it.
772  */
773  if (PQstatus(conn->streamConn) == CONNECTION_BAD)
774  return -1;
775 
776  ereport(ERROR,
777  (errmsg("unexpected result after CommandComplete: %s",
778  PQerrorMessage(conn->streamConn))));
779  }
780 
781  return -1;
782  }
783  else if (PQresultStatus(res) == PGRES_COPY_IN)
784  {
785  PQclear(res);
786  return -1;
787  }
788  else
789  {
790  PQclear(res);
791  ereport(ERROR,
792  (errmsg("could not receive data from WAL stream: %s",
793  pchomp(PQerrorMessage(conn->streamConn)))));
794  }
795  }
796  if (rawlen < -1)
797  ereport(ERROR,
798  (errmsg("could not receive data from WAL stream: %s",
799  pchomp(PQerrorMessage(conn->streamConn)))));
800 
801  /* Return received messages to caller */
802  *buffer = conn->recvBuf;
803  return rawlen;
804 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6672
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2692
int PQgetCopyData(PGconn *conn, char **buffer, int async)
Definition: fe-exec.c:2473
static PGresult * libpqrcv_PQgetResult(PGconn *streamConn)
char * pchomp(const char *in)
Definition: mcxt.c:1214
#define ERROR
Definition: elog.h:43
int PQconsumeInput(PGconn *conn)
Definition: fe-exec.c:1704
#define ereport(elevel,...)
Definition: elog.h:144
void PQclear(PGresult *res)
Definition: fe-exec.c:694
int errmsg(const char *fmt,...)
Definition: elog.c:824
ConnStatusType PQstatus(const PGconn *conn)
Definition: fe-connect.c:6619
void PQfreemem(void *ptr)
Definition: fe-exec.c:3296
int PQsocket(const PGconn *conn)
Definition: fe-connect.c:6690

◆ libpqrcv_send()

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

Definition at line 812 of file libpqwalreceiver.c.

References ereport, errmsg(), ERROR, pchomp(), PQerrorMessage(), PQflush(), PQputCopyData(), and WalReceiverConn::streamConn.

813 {
814  if (PQputCopyData(conn->streamConn, buffer, nbytes) <= 0 ||
815  PQflush(conn->streamConn))
816  ereport(ERROR,
817  (errmsg("could not send data to WAL stream: %s",
818  pchomp(PQerrorMessage(conn->streamConn)))));
819 }
int PQputCopyData(PGconn *conn, const char *buffer, int nbytes)
Definition: fe-exec.c:2317
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6672
char * pchomp(const char *in)
Definition: mcxt.c:1214
#define ERROR
Definition: elog.h:43
int PQflush(PGconn *conn)
Definition: fe-exec.c:3283
#define ereport(elevel,...)
Definition: elog.h:144
int errmsg(const char *fmt,...)
Definition: elog.c:824

◆ libpqrcv_server_version()

static int libpqrcv_server_version ( WalReceiverConn conn)
static

Definition at line 373 of file libpqwalreceiver.c.

References PQserverVersion(), and WalReceiverConn::streamConn.

374 {
375  return PQserverVersion(conn->streamConn);
376 }
int PQserverVersion(const PGconn *conn)
Definition: fe-connect.c:6662

◆ libpqrcv_startstreaming()

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

Definition at line 389 of file libpqwalreceiver.c.

References appendStringInfo(), appendStringInfoChar(), appendStringInfoString(), Assert, StringInfoData::data, ereport, errmsg(), ERROR, initStringInfo(), libpqrcv_PQexec(), WalReceiverConn::logical, WalRcvStreamOptions::logical, pchomp(), pfree(), PGRES_COMMAND_OK, PGRES_COPY_BOTH, WalRcvStreamOptions::physical, PQclear(), PQerrorMessage(), PQescapeLiteral(), PQfreemem(), PQresultStatus(), PQserverVersion(), WalRcvStreamOptions::proto, WalRcvStreamOptions::slotname, WalRcvStreamOptions::startpoint, WalReceiverConn::streamConn, and stringlist_to_identifierstr().

391 {
392  StringInfoData cmd;
393  PGresult *res;
394 
395  Assert(options->logical == conn->logical);
396  Assert(options->slotname || !options->logical);
397 
398  initStringInfo(&cmd);
399 
400  /* Build the command. */
401  appendStringInfoString(&cmd, "START_REPLICATION");
402  if (options->slotname != NULL)
403  appendStringInfo(&cmd, " SLOT \"%s\"",
404  options->slotname);
405 
406  if (options->logical)
407  appendStringInfoString(&cmd, " LOGICAL");
408 
409  appendStringInfo(&cmd, " %X/%X",
410  (uint32) (options->startpoint >> 32),
411  (uint32) options->startpoint);
412 
413  /*
414  * Additional options are different depending on if we are doing logical
415  * or physical replication.
416  */
417  if (options->logical)
418  {
419  char *pubnames_str;
420  List *pubnames;
421  char *pubnames_literal;
422 
423  appendStringInfoString(&cmd, " (");
424 
425  appendStringInfo(&cmd, "proto_version '%u'",
426  options->proto.logical.proto_version);
427 
428  pubnames = options->proto.logical.publication_names;
429  pubnames_str = stringlist_to_identifierstr(conn->streamConn, pubnames);
430  if (!pubnames_str)
431  ereport(ERROR,
432  (errmsg("could not start WAL streaming: %s",
433  pchomp(PQerrorMessage(conn->streamConn)))));
434  pubnames_literal = PQescapeLiteral(conn->streamConn, pubnames_str,
435  strlen(pubnames_str));
436  if (!pubnames_literal)
437  ereport(ERROR,
438  (errmsg("could not start WAL streaming: %s",
439  pchomp(PQerrorMessage(conn->streamConn)))));
440  appendStringInfo(&cmd, ", publication_names %s", pubnames_literal);
441  PQfreemem(pubnames_literal);
442  pfree(pubnames_str);
443 
444  if (options->proto.logical.binary &&
445  PQserverVersion(conn->streamConn) >= 140000)
446  appendStringInfoString(&cmd, ", binary 'true'");
447 
448  appendStringInfoChar(&cmd, ')');
449  }
450  else
451  appendStringInfo(&cmd, " TIMELINE %u",
452  options->proto.physical.startpointTLI);
453 
454  /* Start streaming. */
455  res = libpqrcv_PQexec(conn->streamConn, cmd.data);
456  pfree(cmd.data);
457 
458  if (PQresultStatus(res) == PGRES_COMMAND_OK)
459  {
460  PQclear(res);
461  return false;
462  }
463  else if (PQresultStatus(res) != PGRES_COPY_BOTH)
464  {
465  PQclear(res);
466  ereport(ERROR,
467  (errmsg("could not start WAL streaming: %s",
468  pchomp(PQerrorMessage(conn->streamConn)))));
469  }
470  PQclear(res);
471  return true;
472 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6672
union WalRcvStreamOptions::@104 proto
int PQserverVersion(const PGconn *conn)
Definition: fe-connect.c:6662
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2692
char * pchomp(const char *in)
Definition: mcxt.c:1214
void pfree(void *pointer)
Definition: mcxt.c:1056
void appendStringInfo(StringInfo str, const char *fmt,...)
Definition: stringinfo.c:91
#define ERROR
Definition: elog.h:43
void appendStringInfoString(StringInfo str, const char *s)
Definition: stringinfo.c:176
XLogRecPtr startpoint
Definition: walreceiver.h:168
unsigned int uint32
Definition: c.h:374
void appendStringInfoChar(StringInfo str, char ch)
Definition: stringinfo.c:188
void initStringInfo(StringInfo str)
Definition: stringinfo.c:59
char * PQescapeLiteral(PGconn *conn, const char *str, size_t len)
Definition: fe-exec.c:3564
static PGresult * libpqrcv_PQexec(PGconn *streamConn, const char *query)
#define ereport(elevel,...)
Definition: elog.h:144
void PQclear(PGresult *res)
Definition: fe-exec.c:694
#define Assert(condition)
Definition: c.h:745
static char * stringlist_to_identifierstr(PGconn *conn, List *strings)
int errmsg(const char *fmt,...)
Definition: elog.c:824
struct WalRcvStreamOptions::@104::@105 physical
Definition: pg_list.h:50
void PQfreemem(void *ptr)
Definition: fe-exec.c:3296

◆ stringlist_to_identifierstr()

static char * stringlist_to_identifierstr ( PGconn conn,
List strings 
)
static

Definition at line 1041 of file libpqwalreceiver.c.

References appendStringInfoChar(), appendStringInfoString(), StringInfoData::data, free, initStringInfo(), lfirst, PQescapeIdentifier(), PQfreemem(), strVal, and val.

Referenced by libpqrcv_startstreaming().

1042 {
1043  ListCell *lc;
1044  StringInfoData res;
1045  bool first = true;
1046 
1047  initStringInfo(&res);
1048 
1049  foreach(lc, strings)
1050  {
1051  char *val = strVal(lfirst(lc));
1052  char *val_escaped;
1053 
1054  if (first)
1055  first = false;
1056  else
1057  appendStringInfoChar(&res, ',');
1058 
1059  val_escaped = PQescapeIdentifier(conn, val, strlen(val));
1060  if (!val_escaped)
1061  {
1062  free(res.data);
1063  return NULL;
1064  }
1065  appendStringInfoString(&res, val_escaped);
1066  PQfreemem(val_escaped);
1067  }
1068 
1069  return res.data;
1070 }
#define strVal(v)
Definition: value.h:54
char * PQescapeIdentifier(PGconn *conn, const char *str, size_t len)
Definition: fe-exec.c:3570
void appendStringInfoString(StringInfo str, const char *s)
Definition: stringinfo.c:176
void appendStringInfoChar(StringInfo str, char ch)
Definition: stringinfo.c:188
void initStringInfo(StringInfo str)
Definition: stringinfo.c:59
#define free(a)
Definition: header.h:65
#define lfirst(lc)
Definition: pg_list.h:190
void PQfreemem(void *ptr)
Definition: fe-exec.c:3296
long val
Definition: informix.c:664

Variable Documentation

◆ PG_MODULE_MAGIC

PG_MODULE_MAGIC

Definition at line 37 of file libpqwalreceiver.c.

◆ PQWalReceiverFunctions

WalReceiverFunctionsType PQWalReceiverFunctions
static
Initial value:
= {
}
static bool libpqrcv_startstreaming(WalReceiverConn *conn, const WalRcvStreamOptions *options)
static void libpqrcv_check_conninfo(const char *conninfo)
static int libpqrcv_receive(WalReceiverConn *conn, char **buffer, pgsocket *wait_fd)
static void libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn, TimeLineID tli, char **filename, char **content, int *len)
static char * libpqrcv_identify_system(WalReceiverConn *conn, TimeLineID *primary_tli)
static void libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)
static WalRcvExecResult * libpqrcv_exec(WalReceiverConn *conn, const char *query, const int nRetTypes, const Oid *retTypes)
static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn)
static char * libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname, bool temporary, CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)
static void libpqrcv_disconnect(WalReceiverConn *conn)
static int libpqrcv_server_version(WalReceiverConn *conn)
static WalReceiverConn * libpqrcv_connect(const char *conninfo, bool logical, const char *appname, char **err)
static char * libpqrcv_get_conninfo(WalReceiverConn *conn)
static void libpqrcv_send(WalReceiverConn *conn, const char *buffer, int nbytes)
static void libpqrcv_get_senderinfo(WalReceiverConn *conn, char **sender_host, int *sender_port)

Definition at line 85 of file libpqwalreceiver.c.

Referenced by _PG_init().