PostgreSQL Source Code  git master
 All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros
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 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

void _PG_init ( void  )

Definition at line 104 of file libpqwalreceiver.c.

References elog, ERROR, NULL, PQWalReceiverFunctions, and WalReceiverFunctions.

105 {
106  if (WalReceiverFunctions != NULL)
107  elog(ERROR, "libpqwalreceiver already loaded");
109 }
#define ERROR
Definition: elog.h:43
static WalReceiverFunctionsType PQWalReceiverFunctions
WalReceiverFunctionsType * WalReceiverFunctions
Definition: walreceiver.c:80
#define NULL
Definition: c.h:229
#define elog
Definition: elog.h:219
static void libpqrcv_check_conninfo ( const char *  conninfo)
static

Definition at line 217 of file libpqwalreceiver.c.

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

218 {
219  PQconninfoOption *opts = NULL;
220  char *err = NULL;
221 
222  opts = PQconninfoParse(conninfo, &err);
223  if (opts == NULL)
224  ereport(ERROR,
225  (errcode(ERRCODE_SYNTAX_ERROR),
226  errmsg("invalid connection string syntax: %s", err)));
227 
228  PQconninfoFree(opts);
229 }
int errcode(int sqlerrcode)
Definition: elog.c:575
PQconninfoOption * PQconninfoParse(const char *conninfo, char **errmsg)
Definition: fe-connect.c:4717
#define ERROR
Definition: elog.h:43
void PQconninfoFree(PQconninfoOption *connOptions)
Definition: fe-connect.c:5947
#define ereport(elevel, rest)
Definition: elog.h:122
#define NULL
Definition: c.h:229
int errmsg(const char *fmt,...)
Definition: elog.c:797
static WalReceiverConn * libpqrcv_connect ( const char *  conninfo,
bool  logical,
const char *  appname,
char **  err 
)
static

Definition at line 117 of file libpqwalreceiver.c.

References Assert, CHECK_FOR_INTERRUPTS, conn, CONNECTION_BAD, CONNECTION_OK, GetDatabaseEncodingName(), i, WalReceiverConn::logical, MyLatch, NULL, 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, WaitLatchOrSocket(), WL_LATCH_SET, WL_POSTMASTER_DEATH, WL_SOCKET_READABLE, and WL_SOCKET_WRITEABLE.

119 {
122  const char *keys[5];
123  const char *vals[5];
124  int i = 0;
125 
126  /*
127  * We use the expand_dbname parameter to process the connection string (or
128  * URI), and pass some extra options. The deliberately undocumented
129  * parameter "replication=true" makes it a replication connection. The
130  * database name is ignored by the server in replication mode, but specify
131  * "replication" for .pgpass lookup.
132  */
133  keys[i] = "dbname";
134  vals[i] = conninfo;
135  keys[++i] = "replication";
136  vals[i] = logical ? "database" : "true";
137  if (!logical)
138  {
139  keys[++i] = "dbname";
140  vals[i] = "replication";
141  }
142  keys[++i] = "fallback_application_name";
143  vals[i] = appname;
144  if (logical)
145  {
146  keys[++i] = "client_encoding";
147  vals[i] = GetDatabaseEncodingName();
148  }
149  keys[++i] = NULL;
150  vals[i] = NULL;
151 
152  Assert(i < sizeof(keys));
153 
154  conn = palloc0(sizeof(WalReceiverConn));
155  conn->streamConn = PQconnectStartParams(keys, vals,
156  /* expand_dbname = */ true);
157  if (PQstatus(conn->streamConn) == CONNECTION_BAD)
158  {
159  *err = pchomp(PQerrorMessage(conn->streamConn));
160  return NULL;
161  }
162 
163  /*
164  * Poll connection until we have OK or FAILED status.
165  *
166  * Per spec for PQconnectPoll, first wait till socket is write-ready.
167  */
168  status = PGRES_POLLING_WRITING;
169  do
170  {
171  /* Wait for socket ready and/or other events. */
172  int io_flag;
173  int rc;
174 
175  io_flag = (status == PGRES_POLLING_READING
178 
181  WL_LATCH_SET | io_flag,
182  PQsocket(conn->streamConn),
183  0,
185 
186  /* Emergency bailout? */
187  if (rc & WL_POSTMASTER_DEATH)
188  exit(1);
189 
190  /* Interrupted? */
191  if (rc & WL_LATCH_SET)
192  {
195  }
196 
197  /* If socket is ready, advance the libpq state machine */
198  if (rc & io_flag)
199  status = PQconnectPoll(conn->streamConn);
200  } while (status != PGRES_POLLING_OK && status != PGRES_POLLING_FAILED);
201 
202  if (PQstatus(conn->streamConn) != CONNECTION_OK)
203  {
204  *err = pchomp(PQerrorMessage(conn->streamConn));
205  return NULL;
206  }
207 
208  conn->logical = logical;
209 
210  return conn;
211 }
#define WL_SOCKET_WRITEABLE
Definition: latch.h:126
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6097
#define WL_SOCKET_READABLE
Definition: latch.h:125
void ResetLatch(volatile Latch *latch)
Definition: latch.c:498
char * pchomp(const char *in)
Definition: mcxt.c:1101
PGconn * conn
Definition: streamutil.c:43
#define WL_POSTMASTER_DEATH
Definition: latch.h:128
void * palloc0(Size size)
Definition: mcxt.c:878
PostgresPollingStatusType PQconnectPoll(PGconn *conn)
Definition: fe-connect.c:1997
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
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:1021
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:632
struct Latch * MyLatch
Definition: globals.c:52
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:98
static void static void status(const char *fmt,...) pg_attribute_printf(1
Definition: pg_regress.c:224
ConnStatusType PQstatus(const PGconn *conn)
Definition: fe-connect.c:6044
#define WL_LATCH_SET
Definition: latch.h:124
int PQsocket(const PGconn *conn)
Definition: fe-connect.c:6115
static char * libpqrcv_create_slot ( WalReceiverConn conn,
const char *  slotname,
bool  temporary,
CRSSnapshotAction  snapshot_action,
XLogRecPtr lsn 
)
static

Definition at line 759 of file libpqwalreceiver.c.

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

762 {
763  PGresult *res;
764  StringInfoData cmd;
765  char *snapshot;
766 
767  initStringInfo(&cmd);
768 
769  appendStringInfo(&cmd, "CREATE_REPLICATION_SLOT \"%s\"", slotname);
770 
771  if (temporary)
772  appendStringInfo(&cmd, " TEMPORARY");
773 
774  if (conn->logical)
775  {
776  appendStringInfo(&cmd, " LOGICAL pgoutput");
777  switch (snapshot_action)
778  {
779  case CRS_EXPORT_SNAPSHOT:
780  appendStringInfo(&cmd, " EXPORT_SNAPSHOT");
781  break;
783  appendStringInfo(&cmd, " NOEXPORT_SNAPSHOT");
784  break;
785  case CRS_USE_SNAPSHOT:
786  appendStringInfo(&cmd, " USE_SNAPSHOT");
787  break;
788  }
789  }
790 
791  res = libpqrcv_PQexec(conn->streamConn, cmd.data);
792  pfree(cmd.data);
793 
794  if (PQresultStatus(res) != PGRES_TUPLES_OK)
795  {
796  PQclear(res);
797  ereport(ERROR,
798  (errmsg("could not create replication slot \"%s\": %s",
799  slotname, pchomp(PQerrorMessage(conn->streamConn)))));
800  }
801 
803  CStringGetDatum(PQgetvalue(res, 0, 1))));
804  if (!PQgetisnull(res, 0, 2))
805  snapshot = pstrdup(PQgetvalue(res, 0, 2));
806  else
807  snapshot = NULL;
808 
809  PQclear(res);
810 
811  return snapshot;
812 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6097
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3067
char * pstrdup(const char *in)
Definition: mcxt.c:1077
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2596
Datum pg_lsn_in(PG_FUNCTION_ARGS)
Definition: pg_lsn.c:30
char * pchomp(const char *in)
Definition: mcxt.c:1101
void pfree(void *pointer)
Definition: mcxt.c:950
void appendStringInfo(StringInfo str, const char *fmt,...)
Definition: stringinfo.c:78
#define ERROR
Definition: elog.h:43
#define CStringGetDatum(X)
Definition: postgres.h:584
Datum DirectFunctionCall1Coll(PGFunction func, Oid collation, Datum arg1)
Definition: fmgr.c:707
#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:650
#define NULL
Definition: c.h:229
int errmsg(const char *fmt,...)
Definition: elog.c:797
int PQgetisnull(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3092
static void libpqrcv_disconnect ( WalReceiverConn conn)
static

Definition at line 632 of file libpqwalreceiver.c.

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

633 {
634  PQfinish(conn->streamConn);
635  if (conn->recvBuf != NULL)
636  PQfreemem(conn->recvBuf);
637  pfree(conn);
638 }
void PQfinish(PGconn *conn)
Definition: fe-connect.c:3630
void pfree(void *pointer)
Definition: mcxt.c:950
#define NULL
Definition: c.h:229
void PQfreemem(void *ptr)
Definition: fe-exec.c:3200
static void libpqrcv_endstreaming ( WalReceiverConn conn,
TimeLineID next_tli 
)
static

Definition at line 421 of file libpqwalreceiver.c.

References ereport, errmsg(), ERROR, NULL, 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.

422 {
423  PGresult *res;
424 
425  if (PQputCopyEnd(conn->streamConn, NULL) <= 0 ||
426  PQflush(conn->streamConn))
427  ereport(ERROR,
428  (errmsg("could not send end-of-streaming message to primary: %s",
429  pchomp(PQerrorMessage(conn->streamConn)))));
430 
431  *next_tli = 0;
432 
433  /*
434  * After COPY is finished, we should receive a result set indicating the
435  * next timeline's ID, or just CommandComplete if the server was shut
436  * down.
437  *
438  * If we had not yet received CopyDone from the backend, PGRES_COPY_OUT is
439  * also possible in case we aborted the copy in mid-stream.
440  */
441  res = PQgetResult(conn->streamConn);
442  if (PQresultStatus(res) == PGRES_TUPLES_OK)
443  {
444  /*
445  * Read the next timeline's ID. The server also sends the timeline's
446  * starting point, but it is ignored.
447  */
448  if (PQnfields(res) < 2 || PQntuples(res) != 1)
449  ereport(ERROR,
450  (errmsg("unexpected result set after end-of-streaming")));
451  *next_tli = pg_atoi(PQgetvalue(res, 0, 0), sizeof(uint32), 0);
452  PQclear(res);
453 
454  /* the result set should be followed by CommandComplete */
455  res = PQgetResult(conn->streamConn);
456  }
457  else if (PQresultStatus(res) == PGRES_COPY_OUT)
458  {
459  PQclear(res);
460 
461  /* End the copy */
462  if (PQendcopy(conn->streamConn))
463  ereport(ERROR,
464  (errmsg("error while shutting down streaming COPY: %s",
465  pchomp(PQerrorMessage(conn->streamConn)))));
466 
467  /* CommandComplete should follow */
468  res = PQgetResult(conn->streamConn);
469  }
470 
471  if (PQresultStatus(res) != PGRES_COMMAND_OK)
472  ereport(ERROR,
473  (errmsg("error reading result of streaming command: %s",
474  pchomp(PQerrorMessage(conn->streamConn)))));
475  PQclear(res);
476 
477  /* Verify that there are no more results */
478  res = PQgetResult(conn->streamConn);
479  if (res != NULL)
480  ereport(ERROR,
481  (errmsg("unexpected result after CommandComplete: %s",
482  pchomp(PQerrorMessage(conn->streamConn)))));
483 }
int PQnfields(const PGresult *res)
Definition: fe-exec.c:2681
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6097
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3067
int PQputCopyEnd(PGconn *conn, const char *errormsg)
Definition: fe-exec.c:2288
int PQntuples(const PGresult *res)
Definition: fe-exec.c:2673
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2596
char * pchomp(const char *in)
Definition: mcxt.c:1101
#define ERROR
Definition: elog.h:43
int PQflush(PGconn *conn)
Definition: fe-exec.c:3187
unsigned int uint32
Definition: c.h:268
#define ereport(elevel, rest)
Definition: elog.h:122
void PQclear(PGresult *res)
Definition: fe-exec.c:650
int PQendcopy(PGconn *conn)
Definition: fe-exec.c:2522
#define NULL
Definition: c.h:229
int errmsg(const char *fmt,...)
Definition: elog.c:797
PGresult * PQgetResult(PGconn *conn)
Definition: fe-exec.c:1702
int32 pg_atoi(const char *s, int size, int c)
Definition: numutils.c:37
static WalRcvExecResult * libpqrcv_exec ( WalReceiverConn conn,
const char *  query,
const int  nRetTypes,
const Oid retTypes 
)
static

Definition at line 893 of file libpqwalreceiver.c.

References _, ereport, WalRcvExecResult::err, errcode(), errmsg(), ERROR, InvalidOid, libpqrcv_PQexec(), libpqrcv_processTuples(), MyDatabaseId, NULL, 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.

895 {
896  PGresult *pgres = NULL;
897  WalRcvExecResult *walres = palloc0(sizeof(WalRcvExecResult));
898 
899  if (MyDatabaseId == InvalidOid)
900  ereport(ERROR,
901  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
902  errmsg("the query interface requires a database connection")));
903 
904  pgres = libpqrcv_PQexec(conn->streamConn, query);
905 
906  switch (PQresultStatus(pgres))
907  {
908  case PGRES_SINGLE_TUPLE:
909  case PGRES_TUPLES_OK:
910  walres->status = WALRCV_OK_TUPLES;
911  libpqrcv_processTuples(pgres, walres, nRetTypes, retTypes);
912  break;
913 
914  case PGRES_COPY_IN:
915  walres->status = WALRCV_OK_COPY_IN;
916  break;
917 
918  case PGRES_COPY_OUT:
919  walres->status = WALRCV_OK_COPY_OUT;
920  break;
921 
922  case PGRES_COPY_BOTH:
923  walres->status = WALRCV_OK_COPY_BOTH;
924  break;
925 
926  case PGRES_COMMAND_OK:
927  walres->status = WALRCV_OK_COMMAND;
928  break;
929 
930  /* Empty query is considered error. */
931  case PGRES_EMPTY_QUERY:
932  walres->status = WALRCV_ERROR;
933  walres->err = _("empty query");
934  break;
935 
937  case PGRES_FATAL_ERROR:
938  case PGRES_BAD_RESPONSE:
939  walres->status = WALRCV_ERROR;
940  walres->err = pchomp(PQerrorMessage(conn->streamConn));
941  break;
942  }
943 
944  PQclear(pgres);
945 
946  return walres;
947 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6097
int errcode(int sqlerrcode)
Definition: elog.c:575
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2596
char * pchomp(const char *in)
Definition: mcxt.c:1101
#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:878
Oid MyDatabaseId
Definition: globals.c:77
static PGresult * libpqrcv_PQexec(PGconn *streamConn, const char *query)
#define InvalidOid
Definition: postgres_ext.h:36
void PQclear(PGresult *res)
Definition: fe-exec.c:650
#define NULL
Definition: c.h:229
WalRcvExecStatus status
Definition: walreceiver.h:187
int errmsg(const char *fmt,...)
Definition: elog.c:797
#define _(x)
Definition: elog.c:84
static char * libpqrcv_get_conninfo ( WalReceiverConn conn)
static

Definition at line 236 of file libpqwalreceiver.c.

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

237 {
238  PQconninfoOption *conn_opts;
239  PQconninfoOption *conn_opt;
241  char *retval;
242 
243  Assert(conn->streamConn != NULL);
244 
245  initPQExpBuffer(&buf);
246  conn_opts = PQconninfo(conn->streamConn);
247 
248  if (conn_opts == NULL)
249  ereport(ERROR,
250  (errmsg("could not parse connection string: %s",
251  _("out of memory"))));
252 
253  /* build a clean connection string from pieces */
254  for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++)
255  {
256  bool obfuscate;
257 
258  /* Skip debug and empty options */
259  if (strchr(conn_opt->dispchar, 'D') ||
260  conn_opt->val == NULL ||
261  conn_opt->val[0] == '\0')
262  continue;
263 
264  /* Obfuscate security-sensitive options */
265  obfuscate = strchr(conn_opt->dispchar, '*') != NULL;
266 
267  appendPQExpBuffer(&buf, "%s%s=%s",
268  buf.len == 0 ? "" : " ",
269  conn_opt->keyword,
270  obfuscate ? "********" : conn_opt->val);
271  }
272 
273  PQconninfoFree(conn_opts);
274 
275  retval = PQExpBufferDataBroken(buf) ? NULL : pstrdup(buf.data);
276  termPQExpBuffer(&buf);
277  return retval;
278 }
void termPQExpBuffer(PQExpBuffer str)
Definition: pqexpbuffer.c:128
char * pstrdup(const char *in)
Definition: mcxt.c:1077
#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:66
void PQconninfoFree(PQconninfoOption *connOptions)
Definition: fe-connect.c:5947
PQconninfoOption * PQconninfo(PGconn *conn)
Definition: fe-connect.c:5906
#define ereport(elevel, rest)
Definition: elog.h:122
#define PQExpBufferDataBroken(buf)
Definition: pqexpbuffer.h:67
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
int errmsg(const char *fmt,...)
Definition: elog.c:797
#define _(x)
Definition: elog.c:84
void initPQExpBuffer(PQExpBuffer str)
Definition: pqexpbuffer.c:89
static char * libpqrcv_identify_system ( WalReceiverConn conn,
TimeLineID primary_tli,
int *  server_version 
)
static

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

287 {
288  PGresult *res;
289  char *primary_sysid;
290 
291  /*
292  * Get the system identifier and timeline ID as a DataRow message from the
293  * primary server.
294  */
295  res = libpqrcv_PQexec(conn->streamConn, "IDENTIFY_SYSTEM");
296  if (PQresultStatus(res) != PGRES_TUPLES_OK)
297  {
298  PQclear(res);
299  ereport(ERROR,
300  (errmsg("could not receive database system identifier and timeline ID from "
301  "the primary server: %s",
302  pchomp(PQerrorMessage(conn->streamConn)))));
303  }
304  if (PQnfields(res) < 3 || PQntuples(res) != 1)
305  {
306  int ntuples = PQntuples(res);
307  int nfields = PQnfields(res);
308 
309  PQclear(res);
310  ereport(ERROR,
311  (errmsg("invalid response from primary server"),
312  errdetail("Could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields.",
313  ntuples, nfields, 3, 1)));
314  }
315  primary_sysid = pstrdup(PQgetvalue(res, 0, 0));
316  *primary_tli = pg_atoi(PQgetvalue(res, 0, 1), 4, 0);
317  PQclear(res);
318 
320 
321  return primary_sysid;
322 }
int PQnfields(const PGresult *res)
Definition: fe-exec.c:2681
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6097
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3067
char * pstrdup(const char *in)
Definition: mcxt.c:1077
int PQserverVersion(const PGconn *conn)
Definition: fe-connect.c:6087
int PQntuples(const PGresult *res)
Definition: fe-exec.c:2673
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2596
char * pchomp(const char *in)
Definition: mcxt.c:1101
#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:650
static int server_version
Definition: pg_dumpall.c:82
int errmsg(const char *fmt,...)
Definition: elog.c:797
int32 pg_atoi(const char *s, int size, int c)
Definition: numutils.c:37
static PGresult * libpqrcv_PQexec ( PGconn streamConn,
const char *  query 
)
static

Definition at line 544 of file libpqwalreceiver.c.

References CHECK_FOR_INTERRUPTS, CONNECTION_BAD, MyLatch, NULL, PGRES_COPY_BOTH, PGRES_COPY_IN, PGRES_COPY_OUT, PQclear(), PQconsumeInput(), PQgetResult(), PQisBusy(), PQresultStatus(), PQsendQuery(), PQsocket(), PQstatus(), ResetLatch(), result, WAIT_EVENT_LIBPQWALRECEIVER, 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().

545 {
546  PGresult *result = NULL;
547  PGresult *lastResult = NULL;
548 
549  /*
550  * PQexec() silently discards any prior query results on the connection.
551  * This is not required for this function as it's expected that the caller
552  * (which is this library in all cases) will behave correctly and we don't
553  * have to be backwards compatible with old libpq.
554  */
555 
556  /*
557  * Submit a query. Since we don't use non-blocking mode, this also can
558  * block. But its risk is relatively small, so we ignore that for now.
559  */
560  if (!PQsendQuery(streamConn, query))
561  return NULL;
562 
563  for (;;)
564  {
565  /*
566  * Receive data until PQgetResult is ready to get the result without
567  * blocking.
568  */
569  while (PQisBusy(streamConn))
570  {
571  int rc;
572 
573  /*
574  * We don't need to break down the sleep into smaller increments,
575  * since we'll get interrupted by signals and can either handle
576  * interrupts here or elog(FATAL) within SIGTERM signal handler if
577  * the signal arrives in the middle of establishment of
578  * replication connection.
579  */
582  WL_LATCH_SET,
583  PQsocket(streamConn),
584  0,
586 
587  /* Emergency bailout? */
588  if (rc & WL_POSTMASTER_DEATH)
589  exit(1);
590 
591  /* Interrupted? */
592  if (rc & WL_LATCH_SET)
593  {
596  }
597 
598  /* Consume whatever data is available from the socket */
599  if (PQconsumeInput(streamConn) == 0)
600  {
601  /* trouble; drop whatever we had and return NULL */
602  PQclear(lastResult);
603  return NULL;
604  }
605  }
606 
607  /*
608  * Emulate PQexec()'s behavior of returning the last result when there
609  * are many. We are fine with returning just last error message.
610  */
611  result = PQgetResult(streamConn);
612  if (result == NULL)
613  break; /* query is complete */
614 
615  PQclear(lastResult);
616  lastResult = result;
617 
618  if (PQresultStatus(lastResult) == PGRES_COPY_IN ||
619  PQresultStatus(lastResult) == PGRES_COPY_OUT ||
620  PQresultStatus(lastResult) == PGRES_COPY_BOTH ||
621  PQstatus(streamConn) == CONNECTION_BAD)
622  break;
623  }
624 
625  return lastResult;
626 }
return result
Definition: formatting.c:1633
#define WL_SOCKET_READABLE
Definition: latch.h:125
void ResetLatch(volatile Latch *latch)
Definition: latch.c:498
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2596
int PQsendQuery(PGconn *conn, const char *query)
Definition: fe-exec.c:1132
#define WL_POSTMASTER_DEATH
Definition: latch.h:128
int PQconsumeInput(PGconn *conn)
Definition: fe-exec.c:1631
void PQclear(PGresult *res)
Definition: fe-exec.c:650
#define NULL
Definition: c.h:229
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:1681
struct Latch * MyLatch
Definition: globals.c:52
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:98
ConnStatusType PQstatus(const PGconn *conn)
Definition: fe-connect.c:6044
#define WL_LATCH_SET
Definition: latch.h:124
int PQsocket(const PGconn *conn)
Definition: fe-connect.c:6115
PGresult * PQgetResult(PGconn *conn)
Definition: fe-exec.c:1702
static void libpqrcv_processTuples ( PGresult pgres,
WalRcvExecResult walres,
const int  nRetTypes,
const Oid retTypes 
)
static

Definition at line 818 of file libpqwalreceiver.c.

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

Referenced by libpqrcv_exec().

820 {
821  int tupn;
822  int coln;
823  int nfields = PQnfields(pgres);
824  HeapTuple tuple;
825  AttInMetadata *attinmeta;
826  MemoryContext rowcontext;
827  MemoryContext oldcontext;
828 
829  /* Make sure we got expected number of fields. */
830  if (nfields != nRetTypes)
831  ereport(ERROR,
832  (errmsg("invalid query response"),
833  errdetail("Expected %d fields, got %d fields.",
834  nRetTypes, nfields)));
835 
836  walres->tuplestore = tuplestore_begin_heap(true, false, work_mem);
837 
838  /* Create tuple descriptor corresponding to expected result. */
839  walres->tupledesc = CreateTemplateTupleDesc(nRetTypes, false);
840  for (coln = 0; coln < nRetTypes; coln++)
841  TupleDescInitEntry(walres->tupledesc, (AttrNumber) coln + 1,
842  PQfname(pgres, coln), retTypes[coln], -1, 0);
843  attinmeta = TupleDescGetAttInMetadata(walres->tupledesc);
844 
845  /* No point in doing more here if there were no tuples returned. */
846  if (PQntuples(pgres) == 0)
847  return;
848 
849  /* Create temporary context for local allocations. */
851  "libpqrcv query result context",
853 
854  /* Process returned rows. */
855  for (tupn = 0; tupn < PQntuples(pgres); tupn++)
856  {
857  char *cstrs[MaxTupleAttributeNumber];
858 
860 
861  /* Do the allocations in temporary context. */
862  oldcontext = MemoryContextSwitchTo(rowcontext);
863 
864  /*
865  * Fill cstrs with null-terminated strings of column values.
866  */
867  for (coln = 0; coln < nfields; coln++)
868  {
869  if (PQgetisnull(pgres, tupn, coln))
870  cstrs[coln] = NULL;
871  else
872  cstrs[coln] = PQgetvalue(pgres, tupn, coln);
873  }
874 
875  /* Convert row to a tuple, and add it to the tuplestore */
876  tuple = BuildTupleFromCStrings(attinmeta, cstrs);
877  tuplestore_puttuple(walres->tuplestore, tuple);
878 
879  /* Clean up */
880  MemoryContextSwitchTo(oldcontext);
881  MemoryContextReset(rowcontext);
882  }
883 
884  MemoryContextDelete(rowcontext);
885 }
int PQnfields(const PGresult *res)
Definition: fe-exec.c:2681
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:200
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3067
#define MaxTupleAttributeNumber
Definition: htup_details.h:33
char * PQfname(const PGresult *res, int field_num)
Definition: fe-exec.c:2759
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:109
void MemoryContextReset(MemoryContext context)
Definition: mcxt.c:135
int PQntuples(const PGresult *res)
Definition: fe-exec.c:2673
HeapTuple BuildTupleFromCStrings(AttInMetadata *attinmeta, char **values)
Definition: execTuples.c:1115
TupleDesc tupledesc
Definition: walreceiver.h:190
#define ERROR
Definition: elog.h:43
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:165
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:37
void TupleDescInitEntry(TupleDesc desc, AttrNumber attributeNumber, const char *attributeName, Oid oidtypeid, int32 typmod, int attdim)
Definition: tupdesc.c:497
#define ereport(elevel, rest)
Definition: elog.h:122
Tuplestorestate * tuplestore_begin_heap(bool randomAccess, bool interXact, int maxKBytes)
Definition: tuplestore.c:318
MemoryContext AllocSetContextCreate(MemoryContext parent, const char *name, Size minContextSize, Size initBlockSize, Size maxBlockSize)
Definition: aset.c:322
AttInMetadata * TupleDescGetAttInMetadata(TupleDesc tupdesc)
Definition: execTuples.c:1068
int work_mem
Definition: globals.c:113
Tuplestorestate * tuplestore
Definition: walreceiver.h:189
#define NULL
Definition: c.h:229
TupleDesc CreateTemplateTupleDesc(int natts, bool hasoid)
Definition: tupdesc.c:41
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:3092
int16 AttrNumber
Definition: attnum.h:21
static void libpqrcv_readtimelinehistoryfile ( WalReceiverConn conn,
TimeLineID  tli,
char **  filename,
char **  content,
int *  len 
)
static

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

492 {
493  PGresult *res;
494  char cmd[64];
495 
496  Assert(!conn->logical);
497 
498  /*
499  * Request the primary to send over the history file for given timeline.
500  */
501  snprintf(cmd, sizeof(cmd), "TIMELINE_HISTORY %u", tli);
502  res = libpqrcv_PQexec(conn->streamConn, cmd);
503  if (PQresultStatus(res) != PGRES_TUPLES_OK)
504  {
505  PQclear(res);
506  ereport(ERROR,
507  (errmsg("could not receive timeline history file from "
508  "the primary server: %s",
509  pchomp(PQerrorMessage(conn->streamConn)))));
510  }
511  if (PQnfields(res) != 2 || PQntuples(res) != 1)
512  {
513  int ntuples = PQntuples(res);
514  int nfields = PQnfields(res);
515 
516  PQclear(res);
517  ereport(ERROR,
518  (errmsg("invalid response from primary server"),
519  errdetail("Expected 1 tuple with 2 fields, got %d tuples with %d fields.",
520  ntuples, nfields)));
521  }
522  *filename = pstrdup(PQgetvalue(res, 0, 0));
523 
524  *len = PQgetlength(res, 0, 1);
525  *content = palloc(*len);
526  memcpy(*content, PQgetvalue(res, 0, 1), *len);
527  PQclear(res);
528 }
int PQgetlength(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3078
int PQnfields(const PGresult *res)
Definition: fe-exec.c:2681
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6097
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3067
char * pstrdup(const char *in)
Definition: mcxt.c:1077
int snprintf(char *str, size_t count, const char *fmt,...) pg_attribute_printf(3
int PQntuples(const PGresult *res)
Definition: fe-exec.c:2673
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2596
char * pchomp(const char *in)
Definition: mcxt.c:1101
#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:650
#define Assert(condition)
Definition: c.h:675
static char * filename
Definition: pg_dumpall.c:89
void * palloc(Size size)
Definition: mcxt.c:849
int errmsg(const char *fmt,...)
Definition: elog.c:797
static int libpqrcv_receive ( WalReceiverConn conn,
char **  buffer,
pgsocket wait_fd 
)
static

Definition at line 657 of file libpqwalreceiver.c.

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

659 {
660  int rawlen;
661 
662  if (conn->recvBuf != NULL)
663  PQfreemem(conn->recvBuf);
664  conn->recvBuf = NULL;
665 
666  /* Try to receive a CopyData message */
667  rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
668  if (rawlen == 0)
669  {
670  /* Try consuming some data. */
671  if (PQconsumeInput(conn->streamConn) == 0)
672  ereport(ERROR,
673  (errmsg("could not receive data from WAL stream: %s",
674  pchomp(PQerrorMessage(conn->streamConn)))));
675 
676  /* Now that we've consumed some input, try again */
677  rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
678  if (rawlen == 0)
679  {
680  /* Tell caller to try again when our socket is ready. */
681  *wait_fd = PQsocket(conn->streamConn);
682  return 0;
683  }
684  }
685  if (rawlen == -1) /* end-of-streaming or error */
686  {
687  PGresult *res;
688 
689  res = PQgetResult(conn->streamConn);
690  if (PQresultStatus(res) == PGRES_COMMAND_OK)
691  {
692  PQclear(res);
693 
694  /* Verify that there are no more results. */
695  res = PQgetResult(conn->streamConn);
696  if (res != NULL)
697  {
698  PQclear(res);
699 
700  /*
701  * If the other side closed the connection orderly (otherwise
702  * we'd seen an error, or PGRES_COPY_IN) don't report an error
703  * here, but let callers deal with it.
704  */
705  if (PQstatus(conn->streamConn) == CONNECTION_BAD)
706  return -1;
707 
708  ereport(ERROR,
709  (errmsg("unexpected result after CommandComplete: %s",
710  PQerrorMessage(conn->streamConn))));
711  }
712 
713  return -1;
714  }
715  else if (PQresultStatus(res) == PGRES_COPY_IN)
716  {
717  PQclear(res);
718  return -1;
719  }
720  else
721  {
722  PQclear(res);
723  ereport(ERROR,
724  (errmsg("could not receive data from WAL stream: %s",
725  pchomp(PQerrorMessage(conn->streamConn)))));
726  }
727  }
728  if (rawlen < -1)
729  ereport(ERROR,
730  (errmsg("could not receive data from WAL stream: %s",
731  pchomp(PQerrorMessage(conn->streamConn)))));
732 
733  /* Return received messages to caller */
734  *buffer = conn->recvBuf;
735  return rawlen;
736 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6097
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2596
int PQgetCopyData(PGconn *conn, char **buffer, int async)
Definition: fe-exec.c:2377
char * pchomp(const char *in)
Definition: mcxt.c:1101
#define ERROR
Definition: elog.h:43
#define ereport(elevel, rest)
Definition: elog.h:122
int PQconsumeInput(PGconn *conn)
Definition: fe-exec.c:1631
void PQclear(PGresult *res)
Definition: fe-exec.c:650
#define NULL
Definition: c.h:229
WalTimeSample buffer[LAG_TRACKER_BUFFER_SIZE]
Definition: walsender.c:214
int errmsg(const char *fmt,...)
Definition: elog.c:797
ConnStatusType PQstatus(const PGconn *conn)
Definition: fe-connect.c:6044
void PQfreemem(void *ptr)
Definition: fe-exec.c:3200
int PQsocket(const PGconn *conn)
Definition: fe-connect.c:6115
PGresult * PQgetResult(PGconn *conn)
Definition: fe-exec.c:1702
static void libpqrcv_send ( WalReceiverConn conn,
const char *  buffer,
int  nbytes 
)
static

Definition at line 744 of file libpqwalreceiver.c.

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

745 {
746  if (PQputCopyData(conn->streamConn, buffer, nbytes) <= 0 ||
747  PQflush(conn->streamConn))
748  ereport(ERROR,
749  (errmsg("could not send data to WAL stream: %s",
750  pchomp(PQerrorMessage(conn->streamConn)))));
751 }
int PQputCopyData(PGconn *conn, const char *buffer, int nbytes)
Definition: fe-exec.c:2221
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6097
char * pchomp(const char *in)
Definition: mcxt.c:1101
#define ERROR
Definition: elog.h:43
int PQflush(PGconn *conn)
Definition: fe-exec.c:3187
#define ereport(elevel, rest)
Definition: elog.h:122
WalTimeSample buffer[LAG_TRACKER_BUFFER_SIZE]
Definition: walsender.c:214
int errmsg(const char *fmt,...)
Definition: elog.c:797
static bool libpqrcv_startstreaming ( WalReceiverConn conn,
const WalRcvStreamOptions options 
)
static

Definition at line 335 of file libpqwalreceiver.c.

References appendStringInfo(), appendStringInfoChar(), appendStringInfoString(), Assert, StringInfoData::data, ereport, errmsg(), ERROR, initStringInfo(), libpqrcv_PQexec(), WalReceiverConn::logical, WalRcvStreamOptions::logical, NULL, 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().

337 {
338  StringInfoData cmd;
339  PGresult *res;
340 
341  Assert(options->logical == conn->logical);
342  Assert(options->slotname || !options->logical);
343 
344  initStringInfo(&cmd);
345 
346  /* Build the command. */
347  appendStringInfoString(&cmd, "START_REPLICATION");
348  if (options->slotname != NULL)
349  appendStringInfo(&cmd, " SLOT \"%s\"",
350  options->slotname);
351 
352  if (options->logical)
353  appendStringInfo(&cmd, " LOGICAL");
354 
355  appendStringInfo(&cmd, " %X/%X",
356  (uint32) (options->startpoint >> 32),
357  (uint32) options->startpoint);
358 
359  /*
360  * Additional options are different depending on if we are doing logical
361  * or physical replication.
362  */
363  if (options->logical)
364  {
365  char *pubnames_str;
366  List *pubnames;
367  char *pubnames_literal;
368 
369  appendStringInfoString(&cmd, " (");
370 
371  appendStringInfo(&cmd, "proto_version '%u'",
372  options->proto.logical.proto_version);
373 
374  pubnames = options->proto.logical.publication_names;
375  pubnames_str = stringlist_to_identifierstr(conn->streamConn, pubnames);
376  if (!pubnames_str)
377  ereport(ERROR,
378  (errmsg("could not start WAL streaming: %s",
379  pchomp(PQerrorMessage(conn->streamConn)))));
380  pubnames_literal = PQescapeLiteral(conn->streamConn, pubnames_str,
381  strlen(pubnames_str));
382  if (!pubnames_literal)
383  ereport(ERROR,
384  (errmsg("could not start WAL streaming: %s",
385  pchomp(PQerrorMessage(conn->streamConn)))));
386  appendStringInfo(&cmd, ", publication_names %s", pubnames_literal);
387  PQfreemem(pubnames_literal);
388  pfree(pubnames_str);
389 
390  appendStringInfoChar(&cmd, ')');
391  }
392  else
393  appendStringInfo(&cmd, " TIMELINE %u",
394  options->proto.physical.startpointTLI);
395 
396  /* Start streaming. */
397  res = libpqrcv_PQexec(conn->streamConn, cmd.data);
398  pfree(cmd.data);
399 
400  if (PQresultStatus(res) == PGRES_COMMAND_OK)
401  {
402  PQclear(res);
403  return false;
404  }
405  else if (PQresultStatus(res) != PGRES_COPY_BOTH)
406  {
407  PQclear(res);
408  ereport(ERROR,
409  (errmsg("could not start WAL streaming: %s",
410  pchomp(PQerrorMessage(conn->streamConn)))));
411  }
412  PQclear(res);
413  return true;
414 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6097
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2596
char * pchomp(const char *in)
Definition: mcxt.c:1101
void pfree(void *pointer)
Definition: mcxt.c:950
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:145
unsigned int uint32
Definition: c.h:268
#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
union WalRcvStreamOptions::@97 proto
char * PQescapeLiteral(PGconn *conn, const char *str, size_t len)
Definition: fe-exec.c:3468
static PGresult * libpqrcv_PQexec(PGconn *streamConn, const char *query)
void PQclear(PGresult *res)
Definition: fe-exec.c:650
#define NULL
Definition: c.h:229
#define Assert(condition)
Definition: c.h:675
static char * stringlist_to_identifierstr(PGconn *conn, List *strings)
int errmsg(const char *fmt,...)
Definition: elog.c:797
struct WalRcvStreamOptions::@97::@98 physical
Definition: pg_list.h:45
void PQfreemem(void *ptr)
Definition: fe-exec.c:3200
static char * stringlist_to_identifierstr ( PGconn conn,
List strings 
)
static

Definition at line 958 of file libpqwalreceiver.c.

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

Referenced by libpqrcv_startstreaming().

959 {
960  ListCell *lc;
961  StringInfoData res;
962  bool first = true;
963 
964  initStringInfo(&res);
965 
966  foreach(lc, strings)
967  {
968  char *val = strVal(lfirst(lc));
969  char *val_escaped;
970 
971  if (first)
972  first = false;
973  else
974  appendStringInfoChar(&res, ',');
975 
976  val_escaped = PQescapeIdentifier(conn, val, strlen(val));
977  if (!val_escaped)
978  {
979  free(res.data);
980  return NULL;
981  }
982  appendStringInfoString(&res, val_escaped);
983  PQfreemem(val_escaped);
984  }
985 
986  return res.data;
987 }
#define strVal(v)
Definition: value.h:54
char * PQescapeIdentifier(PGconn *conn, const char *str, size_t len)
Definition: fe-exec.c:3474
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 NULL
Definition: c.h:229
#define lfirst(lc)
Definition: pg_list.h:106
void PQfreemem(void *ptr)
Definition: fe-exec.c:3200
long val
Definition: informix.c:689

Variable Documentation

PG_MODULE_MAGIC

Definition at line 36 of file libpqwalreceiver.c.

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)

Definition at line 81 of file libpqwalreceiver.c.

Referenced by _PG_init().