PostgreSQL Source Code  git master
libpqwalreceiver.c File Reference
#include "postgres.h"
#include <unistd.h>
#include <sys/time.h>
#include "libpq-fe.h"
#include "pqexpbuffer.h"
#include "access/xlog.h"
#include "catalog/pg_type.h"
#include "funcapi.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "pgstat.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, int *server_version)
 
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 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 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 107 of file libpqwalreceiver.c.

References elog, ERROR, PQWalReceiverFunctions, and WalReceiverFunctions.

108 {
109  if (WalReceiverFunctions != NULL)
110  elog(ERROR, "libpqwalreceiver already loaded");
112 }
#define ERROR
Definition: elog.h:43
static WalReceiverFunctionsType PQWalReceiverFunctions
WalReceiverFunctionsType * WalReceiverFunctions
Definition: walreceiver.c:81
#define elog
Definition: elog.h:219

◆ libpqrcv_check_conninfo()

static void libpqrcv_check_conninfo ( const char *  conninfo)
static

Definition at line 226 of file libpqwalreceiver.c.

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

227 {
228  PQconninfoOption *opts = NULL;
229  char *err = NULL;
230 
231  opts = PQconninfoParse(conninfo, &err);
232  if (opts == NULL)
233  ereport(ERROR,
234  (errcode(ERRCODE_SYNTAX_ERROR),
235  errmsg("invalid connection string syntax: %s", err)));
236 
237  PQconninfoFree(opts);
238 }
int errcode(int sqlerrcode)
Definition: elog.c:575
PQconninfoOption * PQconninfoParse(const char *conninfo, char **errmsg)
Definition: fe-connect.c:4735
#define ERROR
Definition: elog.h:43
void PQconninfoFree(PQconninfoOption *connOptions)
Definition: fe-connect.c:5965
#define ereport(elevel, rest)
Definition: elog.h:122
int errmsg(const char *fmt,...)
Definition: elog.c:797

◆ libpqrcv_connect()

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

Definition at line 120 of file libpqwalreceiver.c.

References Assert, CHECK_FOR_INTERRUPTS, conn, CONNECTION_BAD, CONNECTION_OK, CONNECTION_STARTED, GetDatabaseEncodingName(), i, WalReceiverConn::logical, MyLatch, palloc0(), pchomp(), PGRES_POLLING_FAILED, PGRES_POLLING_OK, PGRES_POLLING_READING, PGRES_POLLING_WRITING, PQconnectPoll(), PQconnectStartParams(), PQerrorMessage(), PQsocket(), PQstatus(), ResetLatch(), status(), WalReceiverConn::streamConn, WAIT_EVENT_LIBPQWALRECEIVER_CONNECT, WaitLatchOrSocket(), WL_LATCH_SET, WL_POSTMASTER_DEATH, WL_SOCKET_CONNECTED, WL_SOCKET_READABLE, and WL_SOCKET_WRITEABLE.

122 {
125  const char *keys[5];
126  const char *vals[5];
127  int i = 0;
128 
129  /*
130  * We use the expand_dbname parameter to process the connection string (or
131  * URI), and pass some extra options.
132  */
133  keys[i] = "dbname";
134  vals[i] = conninfo;
135  keys[++i] = "replication";
136  vals[i] = logical ? "database" : "true";
137  if (!logical)
138  {
139  /*
140  * The database name is ignored by the server in replication mode, but
141  * specify "replication" for .pgpass lookup.
142  */
143  keys[++i] = "dbname";
144  vals[i] = "replication";
145  }
146  keys[++i] = "fallback_application_name";
147  vals[i] = appname;
148  if (logical)
149  {
150  keys[++i] = "client_encoding";
151  vals[i] = GetDatabaseEncodingName();
152  }
153  keys[++i] = NULL;
154  vals[i] = NULL;
155 
156  Assert(i < sizeof(keys));
157 
158  conn = palloc0(sizeof(WalReceiverConn));
159  conn->streamConn = PQconnectStartParams(keys, vals,
160  /* expand_dbname = */ true);
161  if (PQstatus(conn->streamConn) == CONNECTION_BAD)
162  {
163  *err = pchomp(PQerrorMessage(conn->streamConn));
164  return NULL;
165  }
166 
167  /*
168  * Poll connection until we have OK or FAILED status.
169  *
170  * Per spec for PQconnectPoll, first wait till socket is write-ready.
171  */
172  status = PGRES_POLLING_WRITING;
173  do
174  {
175  int io_flag;
176  int rc;
177 
178  if (status == PGRES_POLLING_READING)
179  io_flag = WL_SOCKET_READABLE;
180 #ifdef WIN32
181  /* Windows needs a different test while waiting for connection-made */
182  else if (PQstatus(conn->streamConn) == CONNECTION_STARTED)
183  io_flag = WL_SOCKET_CONNECTED;
184 #endif
185  else
186  io_flag = WL_SOCKET_WRITEABLE;
187 
190  WL_LATCH_SET | io_flag,
191  PQsocket(conn->streamConn),
192  0,
194 
195  /* Emergency bailout? */
196  if (rc & WL_POSTMASTER_DEATH)
197  exit(1);
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  conn->logical = logical;
218 
219  return conn;
220 }
#define WL_SOCKET_WRITEABLE
Definition: latch.h:126
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6116
#define WL_SOCKET_READABLE
Definition: latch.h:125
void ResetLatch(volatile Latch *latch)
Definition: latch.c:497
char * pchomp(const char *in)
Definition: mcxt.c:1189
PGconn * conn
Definition: streamutil.c:55
#define WL_POSTMASTER_DEATH
Definition: latch.h:128
void * palloc0(Size size)
Definition: mcxt.c:955
PostgresPollingStatusType PQconnectPoll(PGconn *conn)
Definition: fe-connect.c:2003
#define Assert(condition)
Definition: c.h:699
int WaitLatchOrSocket(volatile Latch *latch, int wakeEvents, pgsocket sock, long timeout, uint32 wait_event_info)
Definition: latch.c:356
const char * GetDatabaseEncodingName(void)
Definition: mbutils.c:1010
PostgresPollingStatusType
Definition: libpq-fe.h:72
int i
PGconn * PQconnectStartParams(const char *const *keywords, const char *const *values, int expand_dbname)
Definition: fe-connect.c:639
struct Latch * MyLatch
Definition: globals.c:55
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:98
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:6063
#define WL_SOCKET_CONNECTED
Definition: latch.h:133
#define WL_LATCH_SET
Definition: latch.h:124
int PQsocket(const PGconn *conn)
Definition: fe-connect.c:6134

◆ libpqrcv_create_slot()

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

Definition at line 791 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.

794 {
795  PGresult *res;
796  StringInfoData cmd;
797  char *snapshot;
798 
799  initStringInfo(&cmd);
800 
801  appendStringInfo(&cmd, "CREATE_REPLICATION_SLOT \"%s\"", slotname);
802 
803  if (temporary)
804  appendStringInfoString(&cmd, " TEMPORARY");
805 
806  if (conn->logical)
807  {
808  appendStringInfoString(&cmd, " LOGICAL pgoutput");
809  switch (snapshot_action)
810  {
811  case CRS_EXPORT_SNAPSHOT:
812  appendStringInfoString(&cmd, " EXPORT_SNAPSHOT");
813  break;
815  appendStringInfoString(&cmd, " NOEXPORT_SNAPSHOT");
816  break;
817  case CRS_USE_SNAPSHOT:
818  appendStringInfoString(&cmd, " USE_SNAPSHOT");
819  break;
820  }
821  }
822 
823  res = libpqrcv_PQexec(conn->streamConn, cmd.data);
824  pfree(cmd.data);
825 
826  if (PQresultStatus(res) != PGRES_TUPLES_OK)
827  {
828  PQclear(res);
829  ereport(ERROR,
830  (errmsg("could not create replication slot \"%s\": %s",
831  slotname, pchomp(PQerrorMessage(conn->streamConn)))));
832  }
833 
835  CStringGetDatum(PQgetvalue(res, 0, 1))));
836  if (!PQgetisnull(res, 0, 2))
837  snapshot = pstrdup(PQgetvalue(res, 0, 2));
838  else
839  snapshot = NULL;
840 
841  PQclear(res);
842 
843  return snapshot;
844 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6116
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3118
char * pstrdup(const char *in)
Definition: mcxt.c:1161
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2647
Datum pg_lsn_in(PG_FUNCTION_ARGS)
Definition: pg_lsn.c:30
char * pchomp(const char *in)
Definition: mcxt.c:1189
void pfree(void *pointer)
Definition: mcxt.c:1031
void appendStringInfo(StringInfo str, const char *fmt,...)
Definition: stringinfo.c:78
#define ERROR
Definition: elog.h:43
void appendStringInfoString(StringInfo str, const char *s)
Definition: stringinfo.c:157
#define CStringGetDatum(X)
Definition: postgres.h:563
Datum DirectFunctionCall1Coll(PGFunction func, Oid collation, Datum arg1)
Definition: fmgr.c:793
#define ereport(elevel, rest)
Definition: elog.h:122
void initStringInfo(StringInfo str)
Definition: stringinfo.c:46
static PGresult * libpqrcv_PQexec(PGconn *streamConn, const char *query)
#define InvalidOid
Definition: postgres_ext.h:36
#define DatumGetLSN(X)
Definition: pg_lsn.h:21
void PQclear(PGresult *res)
Definition: fe-exec.c:671
int errmsg(const char *fmt,...)
Definition: elog.c:797
int PQgetisnull(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3143

◆ libpqrcv_disconnect()

static void libpqrcv_disconnect ( WalReceiverConn conn)
static

Definition at line 664 of file libpqwalreceiver.c.

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

665 {
666  PQfinish(conn->streamConn);
667  if (conn->recvBuf != NULL)
668  PQfreemem(conn->recvBuf);
669  pfree(conn);
670 }
void PQfinish(PGconn *conn)
Definition: fe-connect.c:3638
void pfree(void *pointer)
Definition: mcxt.c:1031
void PQfreemem(void *ptr)
Definition: fe-exec.c:3251

◆ libpqrcv_endstreaming()

static void libpqrcv_endstreaming ( WalReceiverConn conn,
TimeLineID next_tli 
)
static

Definition at line 453 of file libpqwalreceiver.c.

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

454 {
455  PGresult *res;
456 
457  if (PQputCopyEnd(conn->streamConn, NULL) <= 0 ||
458  PQflush(conn->streamConn))
459  ereport(ERROR,
460  (errmsg("could not send end-of-streaming message to primary: %s",
461  pchomp(PQerrorMessage(conn->streamConn)))));
462 
463  *next_tli = 0;
464 
465  /*
466  * After COPY is finished, we should receive a result set indicating the
467  * next timeline's ID, or just CommandComplete if the server was shut
468  * down.
469  *
470  * If we had not yet received CopyDone from the backend, PGRES_COPY_OUT is
471  * also possible in case we aborted the copy in mid-stream.
472  */
473  res = PQgetResult(conn->streamConn);
474  if (PQresultStatus(res) == PGRES_TUPLES_OK)
475  {
476  /*
477  * Read the next timeline's ID. The server also sends the timeline's
478  * starting point, but it is ignored.
479  */
480  if (PQnfields(res) < 2 || PQntuples(res) != 1)
481  ereport(ERROR,
482  (errmsg("unexpected result set after end-of-streaming")));
483  *next_tli = pg_atoi(PQgetvalue(res, 0, 0), sizeof(uint32), 0);
484  PQclear(res);
485 
486  /* the result set should be followed by CommandComplete */
487  res = PQgetResult(conn->streamConn);
488  }
489  else if (PQresultStatus(res) == PGRES_COPY_OUT)
490  {
491  PQclear(res);
492 
493  /* End the copy */
494  if (PQendcopy(conn->streamConn))
495  ereport(ERROR,
496  (errmsg("error while shutting down streaming COPY: %s",
497  pchomp(PQerrorMessage(conn->streamConn)))));
498 
499  /* CommandComplete should follow */
500  res = PQgetResult(conn->streamConn);
501  }
502 
503  if (PQresultStatus(res) != PGRES_COMMAND_OK)
504  ereport(ERROR,
505  (errmsg("error reading result of streaming command: %s",
506  pchomp(PQerrorMessage(conn->streamConn)))));
507  PQclear(res);
508 
509  /* Verify that there are no more results */
510  res = PQgetResult(conn->streamConn);
511  if (res != NULL)
512  ereport(ERROR,
513  (errmsg("unexpected result after CommandComplete: %s",
514  pchomp(PQerrorMessage(conn->streamConn)))));
515 }
int PQnfields(const PGresult *res)
Definition: fe-exec.c:2732
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6116
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3118
int PQputCopyEnd(PGconn *conn, const char *errormsg)
Definition: fe-exec.c:2339
int PQntuples(const PGresult *res)
Definition: fe-exec.c:2724
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2647
char * pchomp(const char *in)
Definition: mcxt.c:1189
#define ERROR
Definition: elog.h:43
int PQflush(PGconn *conn)
Definition: fe-exec.c:3238
unsigned int uint32
Definition: c.h:325
#define ereport(elevel, rest)
Definition: elog.h:122
void PQclear(PGresult *res)
Definition: fe-exec.c:671
int PQendcopy(PGconn *conn)
Definition: fe-exec.c:2573
int errmsg(const char *fmt,...)
Definition: elog.c:797
PGresult * PQgetResult(PGconn *conn)
Definition: fe-exec.c:1753
int32 pg_atoi(const char *s, int size, int c)
Definition: numutils.c:37

◆ libpqrcv_exec()

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

Definition at line 925 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.

927 {
928  PGresult *pgres = NULL;
929  WalRcvExecResult *walres = palloc0(sizeof(WalRcvExecResult));
930 
931  if (MyDatabaseId == InvalidOid)
932  ereport(ERROR,
933  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
934  errmsg("the query interface requires a database connection")));
935 
936  pgres = libpqrcv_PQexec(conn->streamConn, query);
937 
938  switch (PQresultStatus(pgres))
939  {
940  case PGRES_SINGLE_TUPLE:
941  case PGRES_TUPLES_OK:
942  walres->status = WALRCV_OK_TUPLES;
943  libpqrcv_processTuples(pgres, walres, nRetTypes, retTypes);
944  break;
945 
946  case PGRES_COPY_IN:
947  walres->status = WALRCV_OK_COPY_IN;
948  break;
949 
950  case PGRES_COPY_OUT:
951  walres->status = WALRCV_OK_COPY_OUT;
952  break;
953 
954  case PGRES_COPY_BOTH:
955  walres->status = WALRCV_OK_COPY_BOTH;
956  break;
957 
958  case PGRES_COMMAND_OK:
959  walres->status = WALRCV_OK_COMMAND;
960  break;
961 
962  /* Empty query is considered error. */
963  case PGRES_EMPTY_QUERY:
964  walres->status = WALRCV_ERROR;
965  walres->err = _("empty query");
966  break;
967 
969  case PGRES_FATAL_ERROR:
970  case PGRES_BAD_RESPONSE:
971  walres->status = WALRCV_ERROR;
972  walres->err = pchomp(PQerrorMessage(conn->streamConn));
973  break;
974  }
975 
976  PQclear(pgres);
977 
978  return walres;
979 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6116
int errcode(int sqlerrcode)
Definition: elog.c:575
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2647
char * pchomp(const char *in)
Definition: mcxt.c:1189
#define ERROR
Definition: elog.h:43
static void libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres, const int nRetTypes, const Oid *retTypes)
#define ereport(elevel, rest)
Definition: elog.h:122
void * palloc0(Size size)
Definition: mcxt.c:955
Oid MyDatabaseId
Definition: globals.c:86
static PGresult * libpqrcv_PQexec(PGconn *streamConn, const char *query)
#define InvalidOid
Definition: postgres_ext.h:36
void PQclear(PGresult *res)
Definition: fe-exec.c:671
WalRcvExecStatus status
Definition: walreceiver.h:196
int errmsg(const char *fmt,...)
Definition: elog.c:797
#define _(x)
Definition: elog.c:84

◆ libpqrcv_get_conninfo()

static char * libpqrcv_get_conninfo ( WalReceiverConn conn)
static

Definition at line 245 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.

246 {
247  PQconninfoOption *conn_opts;
248  PQconninfoOption *conn_opt;
250  char *retval;
251 
252  Assert(conn->streamConn != NULL);
253 
254  initPQExpBuffer(&buf);
255  conn_opts = PQconninfo(conn->streamConn);
256 
257  if (conn_opts == NULL)
258  ereport(ERROR,
259  (errmsg("could not parse connection string: %s",
260  _("out of memory"))));
261 
262  /* build a clean connection string from pieces */
263  for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++)
264  {
265  bool obfuscate;
266 
267  /* Skip debug and empty options */
268  if (strchr(conn_opt->dispchar, 'D') ||
269  conn_opt->val == NULL ||
270  conn_opt->val[0] == '\0')
271  continue;
272 
273  /* Obfuscate security-sensitive options */
274  obfuscate = strchr(conn_opt->dispchar, '*') != NULL;
275 
276  appendPQExpBuffer(&buf, "%s%s=%s",
277  buf.len == 0 ? "" : " ",
278  conn_opt->keyword,
279  obfuscate ? "********" : conn_opt->val);
280  }
281 
282  PQconninfoFree(conn_opts);
283 
284  retval = PQExpBufferDataBroken(buf) ? NULL : pstrdup(buf.data);
285  termPQExpBuffer(&buf);
286  return retval;
287 }
void termPQExpBuffer(PQExpBuffer str)
Definition: pqexpbuffer.c:128
char * pstrdup(const char *in)
Definition: mcxt.c:1161
#define ERROR
Definition: elog.h:43
void appendPQExpBuffer(PQExpBuffer str, const char *fmt,...)
Definition: pqexpbuffer.c:262
static char * buf
Definition: pg_test_fsync.c:67
void PQconninfoFree(PQconninfoOption *connOptions)
Definition: fe-connect.c:5965
PQconninfoOption * PQconninfo(PGconn *conn)
Definition: fe-connect.c:5924
#define ereport(elevel, rest)
Definition: elog.h:122
#define PQExpBufferDataBroken(buf)
Definition: pqexpbuffer.h:67
#define Assert(condition)
Definition: c.h:699
int errmsg(const char *fmt,...)
Definition: elog.c:797
#define _(x)
Definition: elog.c:84
void initPQExpBuffer(PQExpBuffer str)
Definition: pqexpbuffer.c:89

◆ libpqrcv_get_senderinfo()

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

Definition at line 293 of file libpqwalreceiver.c.

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

295 {
296  char *ret = NULL;
297 
298  *sender_host = NULL;
299  *sender_port = 0;
300 
301  Assert(conn->streamConn != NULL);
302 
303  ret = PQhost(conn->streamConn);
304  if (ret && strlen(ret) != 0)
305  *sender_host = pstrdup(ret);
306 
307  ret = PQport(conn->streamConn);
308  if (ret && strlen(ret) != 0)
309  *sender_port = atoi(ret);
310 }
char * pstrdup(const char *in)
Definition: mcxt.c:1161
char * PQport(const PGconn *conn)
Definition: fe-connect.c:6035
char * PQhost(const PGconn *conn)
Definition: fe-connect.c:6016
#define Assert(condition)
Definition: c.h:699

◆ libpqrcv_identify_system()

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

Definition at line 317 of file libpqwalreceiver.c.

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

319 {
320  PGresult *res;
321  char *primary_sysid;
322 
323  /*
324  * Get the system identifier and timeline ID as a DataRow message from the
325  * primary server.
326  */
327  res = libpqrcv_PQexec(conn->streamConn, "IDENTIFY_SYSTEM");
328  if (PQresultStatus(res) != PGRES_TUPLES_OK)
329  {
330  PQclear(res);
331  ereport(ERROR,
332  (errmsg("could not receive database system identifier and timeline ID from "
333  "the primary server: %s",
334  pchomp(PQerrorMessage(conn->streamConn)))));
335  }
336  if (PQnfields(res) < 3 || PQntuples(res) != 1)
337  {
338  int ntuples = PQntuples(res);
339  int nfields = PQnfields(res);
340 
341  PQclear(res);
342  ereport(ERROR,
343  (errmsg("invalid response from primary server"),
344  errdetail("Could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields.",
345  ntuples, nfields, 3, 1)));
346  }
347  primary_sysid = pstrdup(PQgetvalue(res, 0, 0));
348  *primary_tli = pg_atoi(PQgetvalue(res, 0, 1), 4, 0);
349  PQclear(res);
350 
352 
353  return primary_sysid;
354 }
int PQnfields(const PGresult *res)
Definition: fe-exec.c:2732
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6116
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3118
char * pstrdup(const char *in)
Definition: mcxt.c:1161
int PQserverVersion(const PGconn *conn)
Definition: fe-connect.c:6106
int PQntuples(const PGresult *res)
Definition: fe-exec.c:2724
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2647
char * pchomp(const char *in)
Definition: mcxt.c:1189
#define ERROR
Definition: elog.h:43
int errdetail(const char *fmt,...)
Definition: elog.c:873
#define ereport(elevel, rest)
Definition: elog.h:122
static PGresult * libpqrcv_PQexec(PGconn *streamConn, const char *query)
void PQclear(PGresult *res)
Definition: fe-exec.c:671
static int server_version
Definition: pg_dumpall.c:79
int errmsg(const char *fmt,...)
Definition: elog.c:797
int32 pg_atoi(const char *s, int size, int c)
Definition: numutils.c:37

◆ libpqrcv_PQexec()

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

Definition at line 576 of file libpqwalreceiver.c.

References CHECK_FOR_INTERRUPTS, CONNECTION_BAD, MyLatch, PGRES_COPY_BOTH, PGRES_COPY_IN, PGRES_COPY_OUT, PQclear(), PQconsumeInput(), PQgetResult(), PQisBusy(), PQresultStatus(), PQsendQuery(), PQsocket(), PQstatus(), ResetLatch(), WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE, WaitLatchOrSocket(), WL_LATCH_SET, WL_POSTMASTER_DEATH, and WL_SOCKET_READABLE.

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

577 {
578  PGresult *result = NULL;
579  PGresult *lastResult = NULL;
580 
581  /*
582  * PQexec() silently discards any prior query results on the connection.
583  * This is not required for this function as it's expected that the caller
584  * (which is this library in all cases) will behave correctly and we don't
585  * have to be backwards compatible with old libpq.
586  */
587 
588  /*
589  * Submit a query. Since we don't use non-blocking mode, this also can
590  * block. But its risk is relatively small, so we ignore that for now.
591  */
592  if (!PQsendQuery(streamConn, query))
593  return NULL;
594 
595  for (;;)
596  {
597  /*
598  * Receive data until PQgetResult is ready to get the result without
599  * blocking.
600  */
601  while (PQisBusy(streamConn))
602  {
603  int rc;
604 
605  /*
606  * We don't need to break down the sleep into smaller increments,
607  * since we'll get interrupted by signals and can either handle
608  * interrupts here or elog(FATAL) within SIGTERM signal handler if
609  * the signal arrives in the middle of establishment of
610  * replication connection.
611  */
614  WL_LATCH_SET,
615  PQsocket(streamConn),
616  0,
618 
619  /* Emergency bailout? */
620  if (rc & WL_POSTMASTER_DEATH)
621  exit(1);
622 
623  /* Interrupted? */
624  if (rc & WL_LATCH_SET)
625  {
628  }
629 
630  /* Consume whatever data is available from the socket */
631  if (PQconsumeInput(streamConn) == 0)
632  {
633  /* trouble; drop whatever we had and return NULL */
634  PQclear(lastResult);
635  return NULL;
636  }
637  }
638 
639  /*
640  * Emulate PQexec()'s behavior of returning the last result when there
641  * are many. We are fine with returning just last error message.
642  */
643  result = PQgetResult(streamConn);
644  if (result == NULL)
645  break; /* query is complete */
646 
647  PQclear(lastResult);
648  lastResult = result;
649 
650  if (PQresultStatus(lastResult) == PGRES_COPY_IN ||
651  PQresultStatus(lastResult) == PGRES_COPY_OUT ||
652  PQresultStatus(lastResult) == PGRES_COPY_BOTH ||
653  PQstatus(streamConn) == CONNECTION_BAD)
654  break;
655  }
656 
657  return lastResult;
658 }
#define WL_SOCKET_READABLE
Definition: latch.h:125
void ResetLatch(volatile Latch *latch)
Definition: latch.c:497
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2647
int PQsendQuery(PGconn *conn, const char *query)
Definition: fe-exec.c:1183
#define WL_POSTMASTER_DEATH
Definition: latch.h:128
int PQconsumeInput(PGconn *conn)
Definition: fe-exec.c:1682
void PQclear(PGresult *res)
Definition: fe-exec.c:671
int WaitLatchOrSocket(volatile Latch *latch, int wakeEvents, pgsocket sock, long timeout, uint32 wait_event_info)
Definition: latch.c:356
int PQisBusy(PGconn *conn)
Definition: fe-exec.c:1732
struct Latch * MyLatch
Definition: globals.c:55
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:98
ConnStatusType PQstatus(const PGconn *conn)
Definition: fe-connect.c:6063
#define WL_LATCH_SET
Definition: latch.h:124
int PQsocket(const PGconn *conn)
Definition: fe-connect.c:6134
PGresult * PQgetResult(PGconn *conn)
Definition: fe-exec.c:1753

◆ libpqrcv_processTuples()

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

Definition at line 850 of file libpqwalreceiver.c.

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

Referenced by libpqrcv_exec().

852 {
853  int tupn;
854  int coln;
855  int nfields = PQnfields(pgres);
856  HeapTuple tuple;
857  AttInMetadata *attinmeta;
858  MemoryContext rowcontext;
859  MemoryContext oldcontext;
860 
861  /* Make sure we got expected number of fields. */
862  if (nfields != nRetTypes)
863  ereport(ERROR,
864  (errmsg("invalid query response"),
865  errdetail("Expected %d fields, got %d fields.",
866  nRetTypes, nfields)));
867 
868  walres->tuplestore = tuplestore_begin_heap(true, false, work_mem);
869 
870  /* Create tuple descriptor corresponding to expected result. */
871  walres->tupledesc = CreateTemplateTupleDesc(nRetTypes, false);
872  for (coln = 0; coln < nRetTypes; coln++)
873  TupleDescInitEntry(walres->tupledesc, (AttrNumber) coln + 1,
874  PQfname(pgres, coln), retTypes[coln], -1, 0);
875  attinmeta = TupleDescGetAttInMetadata(walres->tupledesc);
876 
877  /* No point in doing more here if there were no tuples returned. */
878  if (PQntuples(pgres) == 0)
879  return;
880 
881  /* Create temporary context for local allocations. */
883  "libpqrcv query result context",
885 
886  /* Process returned rows. */
887  for (tupn = 0; tupn < PQntuples(pgres); tupn++)
888  {
889  char *cstrs[MaxTupleAttributeNumber];
890 
892 
893  /* Do the allocations in temporary context. */
894  oldcontext = MemoryContextSwitchTo(rowcontext);
895 
896  /*
897  * Fill cstrs with null-terminated strings of column values.
898  */
899  for (coln = 0; coln < nfields; coln++)
900  {
901  if (PQgetisnull(pgres, tupn, coln))
902  cstrs[coln] = NULL;
903  else
904  cstrs[coln] = PQgetvalue(pgres, tupn, coln);
905  }
906 
907  /* Convert row to a tuple, and add it to the tuplestore */
908  tuple = BuildTupleFromCStrings(attinmeta, cstrs);
909  tuplestore_puttuple(walres->tuplestore, tuple);
910 
911  /* Clean up */
912  MemoryContextSwitchTo(oldcontext);
913  MemoryContextReset(rowcontext);
914  }
915 
916  MemoryContextDelete(rowcontext);
917 }
int PQnfields(const PGresult *res)
Definition: fe-exec.c:2732
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:211
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3118
#define MaxTupleAttributeNumber
Definition: htup_details.h:33
char * PQfname(const PGresult *res, int field_num)
Definition: fe-exec.c:2810
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:2724
HeapTuple BuildTupleFromCStrings(AttInMetadata *attinmeta, char **values)
Definition: execTuples.c:1195
TupleDesc tupledesc
Definition: walreceiver.h:199
#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:873
MemoryContext CurrentMemoryContext
Definition: mcxt.c:38
void TupleDescInitEntry(TupleDesc desc, AttrNumber attributeNumber, const char *attributeName, Oid oidtypeid, int32 typmod, int attdim)
Definition: tupdesc.c:600
#define ereport(elevel, rest)
Definition: elog.h:122
#define AllocSetContextCreate(parent, name, allocparams)
Definition: memutils.h:170
Tuplestorestate * tuplestore_begin_heap(bool randomAccess, bool interXact, int maxKBytes)
Definition: tuplestore.c:318
AttInMetadata * TupleDescGetAttInMetadata(TupleDesc tupdesc)
Definition: execTuples.c:1146
int work_mem
Definition: globals.c:122
Tuplestorestate * tuplestore
Definition: walreceiver.h:198
TupleDesc CreateTemplateTupleDesc(int natts, bool hasoid)
Definition: tupdesc.c:45
int errmsg(const char *fmt,...)
Definition: elog.c:797
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:98
int PQgetisnull(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3143
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 521 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.

524 {
525  PGresult *res;
526  char cmd[64];
527 
528  Assert(!conn->logical);
529 
530  /*
531  * Request the primary to send over the history file for given timeline.
532  */
533  snprintf(cmd, sizeof(cmd), "TIMELINE_HISTORY %u", tli);
534  res = libpqrcv_PQexec(conn->streamConn, cmd);
535  if (PQresultStatus(res) != PGRES_TUPLES_OK)
536  {
537  PQclear(res);
538  ereport(ERROR,
539  (errmsg("could not receive timeline history file from "
540  "the primary server: %s",
541  pchomp(PQerrorMessage(conn->streamConn)))));
542  }
543  if (PQnfields(res) != 2 || PQntuples(res) != 1)
544  {
545  int ntuples = PQntuples(res);
546  int nfields = PQnfields(res);
547 
548  PQclear(res);
549  ereport(ERROR,
550  (errmsg("invalid response from primary server"),
551  errdetail("Expected 1 tuple with 2 fields, got %d tuples with %d fields.",
552  ntuples, nfields)));
553  }
554  *filename = pstrdup(PQgetvalue(res, 0, 0));
555 
556  *len = PQgetlength(res, 0, 1);
557  *content = palloc(*len);
558  memcpy(*content, PQgetvalue(res, 0, 1), *len);
559  PQclear(res);
560 }
int PQgetlength(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3129
int PQnfields(const PGresult *res)
Definition: fe-exec.c:2732
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6116
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3118
char * pstrdup(const char *in)
Definition: mcxt.c:1161
int snprintf(char *str, size_t count, const char *fmt,...) pg_attribute_printf(3
int PQntuples(const PGresult *res)
Definition: fe-exec.c:2724
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2647
char * pchomp(const char *in)
Definition: mcxt.c:1189
#define ERROR
Definition: elog.h:43
int errdetail(const char *fmt,...)
Definition: elog.c:873
#define ereport(elevel, rest)
Definition: elog.h:122
static PGresult * libpqrcv_PQexec(PGconn *streamConn, const char *query)
void PQclear(PGresult *res)
Definition: fe-exec.c:671
#define Assert(condition)
Definition: c.h:699
static char * filename
Definition: pg_dumpall.c:87
void * palloc(Size size)
Definition: mcxt.c:924
int errmsg(const char *fmt,...)
Definition: elog.c:797

◆ libpqrcv_receive()

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

Definition at line 689 of file libpqwalreceiver.c.

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

691 {
692  int rawlen;
693 
694  if (conn->recvBuf != NULL)
695  PQfreemem(conn->recvBuf);
696  conn->recvBuf = NULL;
697 
698  /* Try to receive a CopyData message */
699  rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
700  if (rawlen == 0)
701  {
702  /* Try consuming some data. */
703  if (PQconsumeInput(conn->streamConn) == 0)
704  ereport(ERROR,
705  (errmsg("could not receive data from WAL stream: %s",
706  pchomp(PQerrorMessage(conn->streamConn)))));
707 
708  /* Now that we've consumed some input, try again */
709  rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
710  if (rawlen == 0)
711  {
712  /* Tell caller to try again when our socket is ready. */
713  *wait_fd = PQsocket(conn->streamConn);
714  return 0;
715  }
716  }
717  if (rawlen == -1) /* end-of-streaming or error */
718  {
719  PGresult *res;
720 
721  res = PQgetResult(conn->streamConn);
722  if (PQresultStatus(res) == PGRES_COMMAND_OK)
723  {
724  PQclear(res);
725 
726  /* Verify that there are no more results. */
727  res = PQgetResult(conn->streamConn);
728  if (res != NULL)
729  {
730  PQclear(res);
731 
732  /*
733  * If the other side closed the connection orderly (otherwise
734  * we'd seen an error, or PGRES_COPY_IN) don't report an error
735  * here, but let callers deal with it.
736  */
737  if (PQstatus(conn->streamConn) == CONNECTION_BAD)
738  return -1;
739 
740  ereport(ERROR,
741  (errmsg("unexpected result after CommandComplete: %s",
742  PQerrorMessage(conn->streamConn))));
743  }
744 
745  return -1;
746  }
747  else if (PQresultStatus(res) == PGRES_COPY_IN)
748  {
749  PQclear(res);
750  return -1;
751  }
752  else
753  {
754  PQclear(res);
755  ereport(ERROR,
756  (errmsg("could not receive data from WAL stream: %s",
757  pchomp(PQerrorMessage(conn->streamConn)))));
758  }
759  }
760  if (rawlen < -1)
761  ereport(ERROR,
762  (errmsg("could not receive data from WAL stream: %s",
763  pchomp(PQerrorMessage(conn->streamConn)))));
764 
765  /* Return received messages to caller */
766  *buffer = conn->recvBuf;
767  return rawlen;
768 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6116
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2647
int PQgetCopyData(PGconn *conn, char **buffer, int async)
Definition: fe-exec.c:2428
char * pchomp(const char *in)
Definition: mcxt.c:1189
#define ERROR
Definition: elog.h:43
#define ereport(elevel, rest)
Definition: elog.h:122
int PQconsumeInput(PGconn *conn)
Definition: fe-exec.c:1682
void PQclear(PGresult *res)
Definition: fe-exec.c:671
WalTimeSample buffer[LAG_TRACKER_BUFFER_SIZE]
Definition: walsender.c:215
int errmsg(const char *fmt,...)
Definition: elog.c:797
ConnStatusType PQstatus(const PGconn *conn)
Definition: fe-connect.c:6063
void PQfreemem(void *ptr)
Definition: fe-exec.c:3251
int PQsocket(const PGconn *conn)
Definition: fe-connect.c:6134
PGresult * PQgetResult(PGconn *conn)
Definition: fe-exec.c:1753

◆ libpqrcv_send()

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

Definition at line 776 of file libpqwalreceiver.c.

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

777 {
778  if (PQputCopyData(conn->streamConn, buffer, nbytes) <= 0 ||
779  PQflush(conn->streamConn))
780  ereport(ERROR,
781  (errmsg("could not send data to WAL stream: %s",
782  pchomp(PQerrorMessage(conn->streamConn)))));
783 }
int PQputCopyData(PGconn *conn, const char *buffer, int nbytes)
Definition: fe-exec.c:2272
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6116
char * pchomp(const char *in)
Definition: mcxt.c:1189
#define ERROR
Definition: elog.h:43
int PQflush(PGconn *conn)
Definition: fe-exec.c:3238
#define ereport(elevel, rest)
Definition: elog.h:122
WalTimeSample buffer[LAG_TRACKER_BUFFER_SIZE]
Definition: walsender.c:215
int errmsg(const char *fmt,...)
Definition: elog.c:797

◆ libpqrcv_startstreaming()

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

Definition at line 367 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(), WalRcvStreamOptions::proto, WalRcvStreamOptions::slotname, WalRcvStreamOptions::startpoint, WalReceiverConn::streamConn, and stringlist_to_identifierstr().

369 {
370  StringInfoData cmd;
371  PGresult *res;
372 
373  Assert(options->logical == conn->logical);
374  Assert(options->slotname || !options->logical);
375 
376  initStringInfo(&cmd);
377 
378  /* Build the command. */
379  appendStringInfoString(&cmd, "START_REPLICATION");
380  if (options->slotname != NULL)
381  appendStringInfo(&cmd, " SLOT \"%s\"",
382  options->slotname);
383 
384  if (options->logical)
385  appendStringInfoString(&cmd, " LOGICAL");
386 
387  appendStringInfo(&cmd, " %X/%X",
388  (uint32) (options->startpoint >> 32),
389  (uint32) options->startpoint);
390 
391  /*
392  * Additional options are different depending on if we are doing logical
393  * or physical replication.
394  */
395  if (options->logical)
396  {
397  char *pubnames_str;
398  List *pubnames;
399  char *pubnames_literal;
400 
401  appendStringInfoString(&cmd, " (");
402 
403  appendStringInfo(&cmd, "proto_version '%u'",
404  options->proto.logical.proto_version);
405 
406  pubnames = options->proto.logical.publication_names;
407  pubnames_str = stringlist_to_identifierstr(conn->streamConn, pubnames);
408  if (!pubnames_str)
409  ereport(ERROR,
410  (errmsg("could not start WAL streaming: %s",
411  pchomp(PQerrorMessage(conn->streamConn)))));
412  pubnames_literal = PQescapeLiteral(conn->streamConn, pubnames_str,
413  strlen(pubnames_str));
414  if (!pubnames_literal)
415  ereport(ERROR,
416  (errmsg("could not start WAL streaming: %s",
417  pchomp(PQerrorMessage(conn->streamConn)))));
418  appendStringInfo(&cmd, ", publication_names %s", pubnames_literal);
419  PQfreemem(pubnames_literal);
420  pfree(pubnames_str);
421 
422  appendStringInfoChar(&cmd, ')');
423  }
424  else
425  appendStringInfo(&cmd, " TIMELINE %u",
426  options->proto.physical.startpointTLI);
427 
428  /* Start streaming. */
429  res = libpqrcv_PQexec(conn->streamConn, cmd.data);
430  pfree(cmd.data);
431 
432  if (PQresultStatus(res) == PGRES_COMMAND_OK)
433  {
434  PQclear(res);
435  return false;
436  }
437  else if (PQresultStatus(res) != PGRES_COPY_BOTH)
438  {
439  PQclear(res);
440  ereport(ERROR,
441  (errmsg("could not start WAL streaming: %s",
442  pchomp(PQerrorMessage(conn->streamConn)))));
443  }
444  PQclear(res);
445  return true;
446 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6116
struct WalRcvStreamOptions::@107::@108 physical
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2647
union WalRcvStreamOptions::@107 proto
char * pchomp(const char *in)
Definition: mcxt.c:1189
void pfree(void *pointer)
Definition: mcxt.c:1031
void appendStringInfo(StringInfo str, const char *fmt,...)
Definition: stringinfo.c:78
#define ERROR
Definition: elog.h:43
void appendStringInfoString(StringInfo str, const char *s)
Definition: stringinfo.c:157
XLogRecPtr startpoint
Definition: walreceiver.h:154
unsigned int uint32
Definition: c.h:325
#define ereport(elevel, rest)
Definition: elog.h:122
void appendStringInfoChar(StringInfo str, char ch)
Definition: stringinfo.c:169
void initStringInfo(StringInfo str)
Definition: stringinfo.c:46
char * PQescapeLiteral(PGconn *conn, const char *str, size_t len)
Definition: fe-exec.c:3519
static PGresult * libpqrcv_PQexec(PGconn *streamConn, const char *query)
void PQclear(PGresult *res)
Definition: fe-exec.c:671
#define Assert(condition)
Definition: c.h:699
static char * stringlist_to_identifierstr(PGconn *conn, List *strings)
int errmsg(const char *fmt,...)
Definition: elog.c:797
Definition: pg_list.h:45
void PQfreemem(void *ptr)
Definition: fe-exec.c:3251

◆ stringlist_to_identifierstr()

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

Definition at line 990 of file libpqwalreceiver.c.

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

Referenced by libpqrcv_startstreaming().

991 {
992  ListCell *lc;
993  StringInfoData res;
994  bool first = true;
995 
996  initStringInfo(&res);
997 
998  foreach(lc, strings)
999  {
1000  char *val = strVal(lfirst(lc));
1001  char *val_escaped;
1002 
1003  if (first)
1004  first = false;
1005  else
1006  appendStringInfoChar(&res, ',');
1007 
1008  val_escaped = PQescapeIdentifier(conn, val, strlen(val));
1009  if (!val_escaped)
1010  {
1011  free(res.data);
1012  return NULL;
1013  }
1014  appendStringInfoString(&res, val_escaped);
1015  PQfreemem(val_escaped);
1016  }
1017 
1018  return res.data;
1019 }
#define strVal(v)
Definition: value.h:54
char * PQescapeIdentifier(PGconn *conn, const char *str, size_t len)
Definition: fe-exec.c:3525
void appendStringInfoString(StringInfo str, const char *s)
Definition: stringinfo.c:157
void appendStringInfoChar(StringInfo str, char ch)
Definition: stringinfo.c:169
void initStringInfo(StringInfo str)
Definition: stringinfo.c:46
#define free(a)
Definition: header.h:65
#define lfirst(lc)
Definition: pg_list.h:106
void PQfreemem(void *ptr)
Definition: fe-exec.c:3251
long val
Definition: informix.c:689

Variable Documentation

◆ PG_MODULE_MAGIC

PG_MODULE_MAGIC

Definition at line 36 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 void libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)
static WalRcvExecResult * libpqrcv_exec(WalReceiverConn *conn, const char *query, const int nRetTypes, const Oid *retTypes)
static char * libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname, bool temporary, CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)
static char * libpqrcv_identify_system(WalReceiverConn *conn, TimeLineID *primary_tli, int *server_version)
static void libpqrcv_disconnect(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 83 of file libpqwalreceiver.c.

Referenced by _PG_init().