Function CopyStreamReceive

src/bin/pg_basebackup/receivelog.c:935–985

* Receive CopyData message available from XLOG stream, blocking for * maximum of 'timeout' ms. * * If data was received, returns the length of the data. *buffer is set to * point to a buffer holding the received message. The buffer is only valid * until the next CopyStreamReceive call. * * Returns 0 if no data was available within timeout, or if wait was * interrupted by signal or stop_soc

 * -1 on error. -2 if the server ended the COPY.
 */
static int
CopyStreamReceive(PGconn *conn, long timeout, pgsocket stop_socket,
				  char **buffer)
{
	char	   *copybuf = NULL;
	int			rawlen;

	if (*buffer != NULL)
		PQfreemem(*buffer);
	*buffer = NULL;

	/* Try to receive a CopyData message */
	rawlen = PQgetCopyData(conn, &copybuf, 1);
	if (rawlen == 0)
	{
		int			ret;

		/*
		 * No data available.  Wait for some to appear, but not longer than
		 * the specified timeout, so that we can ping the server.  Also stop
		 * waiting if input appears on stop_socket.
		 */
		ret = CopyStreamPoll(conn, timeout, stop_socket);
		if (ret <= 0)
			return ret;

		/* Now there is actually data on the socket */
		if (PQconsumeInput(conn) == 0)
		{
			pg_log_error("could not receive data from WAL stream: %s",
						 PQerrorMessage(conn));
			return -1;
		}

		/* Now that we've consumed some input, try again */
		rawlen = PQgetCopyData(conn, &copybuf, 1);
		if (rawlen == 0)
			return 0;
	}
	if (rawlen == -1)			/* end-of-streaming or error */
		return -2;
	if (rawlen == -2)
	{
		pg_log_error("could not read COPY data: %s", PQerrorMessage(conn));
		return -1;
	}

	/* Return received messages to caller */
	*buffer = copybuf;
	return rawlen;
}

/*
 * Process the keepalive message.