/*--------------------------------------------------------------------
 * FILE:
 *     recovery.c
 *
 * NOTE:
 *     This file is composed of the functions to call with the source
 *     at backend for the recovery.
 *     Low level I/O functions that called by in these functions are 
 *     contained in 'replicate_com.c'.
 *
 *--------------------------------------------------------------------
 */

/*--------------------------------------
 * INTERFACE ROUTINES
 *
 * I/O call:
 *      PGR_recovery_finish_send
 * master module:
 *      PGR_Master_Main(void);
 * recovery module:
 *      PGR_Recovery_Main
 *-------------------------------------
 */
#ifdef USE_REPLICATION

#include "postgres.h"

#include <stdio.h>
#include <unistd.h>
#include <signal.h>
#include <sys/wait.h>
#include <ctype.h>
#include <time.h>
#include <pwd.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/socket.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <netdb.h>
#include <netinet/in.h>
#include <errno.h>
#include <fcntl.h>
#include <time.h>
#include <sys/param.h>
#include <sys/select.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <sys/file.h>
#include <dirent.h>

#include "libpq/pqsignal.h"
#include "utils/guc.h"
#include "miscadmin.h"
#include "nodes/nodes.h"
#include "nodes/parsenodes.h"
#include "access/xact.h"
#include "access/xlog.h"
#include "tcop/tcopprot.h"

#include "replicate.h"

#ifdef WIN32
#include "win32.h"
#else
#ifdef HAVE_NETINET_TCP_H
#include <netinet/tcp.h>
#endif
#include <arpa/inet.h>
#endif

#ifndef HAVE_STRDUP
#include "strdup.h"
#endif
#ifdef HAVE_CRYPT_H
#include <crypt.h>
#endif

#ifdef MULTIBYTE
#include "mb/pg_wchar.h"
#endif

#define RECOVERY_LOOP_END	(0)
#define RECOVERY_LOOP_CONTINUE	(1)
#define RECOVERY_LOOP_FAIL	(2)
char Local_Host_Name[HOSTNAME_MAX_LENGTH];

static int read_packet(int sock,RecoveryPacket * packet);
static int send_recovery_packet(int  sock, RecoveryPacket * packet);
static int send_packet(int * sock, RecoveryPacket * packet );
static void master_loop(int fd);
static int start_recovery_send(int * sock, ReplicateServerInfo * host);
static int stop_recovery_send(int * sock, ReplicateServerInfo * host);
static int rsync_pg_data(char * src , char * dest);
static int remove_dir(char * dir_name);
static int clear_bkup_dir(char * dir_name);
static int bkup_dir(char * dir_name);
static int restore_dir(char * dir_name);
static int rsync_global_dir(char * src, char * dest);
static int first_recovery(char * src, char * dest, char * dir);
static int second_recovery(char * src, char * dest, char * dir);
static int recovery_rsync(char * src , char * dest, int stage);
static int recovery_loop(int fd);
static void show_recovery_packet(RecoveryPacket * packet);
static int direct_send_packet(int packet_no);
static void set_recovery_packet(RecoveryPacket * packet, int packet_no);

int PGR_recovery_error_send(void);
int PGR_recovery_finish_send(void);
int PGR_Master_Main(void);
int PGR_Recovery_Main(void);

static int
read_packet(int sock,RecoveryPacket * packet)
{
	int r;
	char * read_ptr;
	int read_size = 0;
	int packet_size = 0;

	read_ptr = (char*)packet;
	packet_size = sizeof(RecoveryPacket);

	for (;;){
		r = recv(sock,read_ptr + read_size ,packet_size, MSG_WAITALL);
		if (r < 0) {
			if (errno == EINTR || errno == EAGAIN) {
				continue;
			} else {
				elog(DEBUG1, "read_packet():recv failed");
				return -1;
			}
		} else if (r == 0) {
			elog(DEBUG1, "read_packet():unexpected EOF");
			return -1;
		} else /*if (r > 0)*/ {
			read_size += r;
			if (read_size == packet_size) {
				show_recovery_packet(packet);
				return read_size;
			}
		}
	}
	return -1;
}

static int
send_recovery_packet(int  sock, RecoveryPacket * packet)
{
	char * send_ptr;
	int send_size= 0;
	int buf_size = 0;
	int s;
	int rtn;	
	fd_set	  wmask;
	struct timeval timeout;

	timeout.tv_sec = PGR_SEND_TIMEOUT;
	timeout.tv_usec = 0;

	/*
	 * Wait for something to happen.
	 */
	rtn = 1;
	while (rtn)
	{
		for (;;)
		{
			timeout.tv_sec = PGR_SEND_TIMEOUT;
			timeout.tv_usec = 0;

			FD_ZERO(&wmask);
			FD_SET(sock,&wmask);
			rtn = select(sock+1, (fd_set *)NULL, &wmask, (fd_set *)NULL, &timeout);

			if (rtn < 0)
			{
				if (errno == EINTR || errno == EAGAIN)
				{
					continue;
				}
				else
				{
					rtn = 0;
					break;
				}
			}
			else if (rtn && FD_ISSET(sock, &wmask))
			{
				send_ptr = (char *)packet;
				buf_size = sizeof(RecoveryPacket);

				s = send(sock,send_ptr + send_size,buf_size - send_size ,0);
				if (s < 0) {
					if (errno == EINTR || errno == EAGAIN) {
						continue;
					}
					elog(DEBUG1, "send_recovery_packet():send error");

					/* EPIPE || ENCONNREFUSED || ENSOCK || EHOSTUNREACH */
					return STATUS_ERROR;
				} else if (s == 0) {
					elog(DEBUG1, "send_recovery_packet():unexpected EOF");
					return STATUS_ERROR;
				} else /*if (s > 0)*/ {
					send_size += s;
					if (send_size == buf_size)
					{
						return STATUS_OK;
					}
				}
			}
		}
	}
	return STATUS_ERROR;
}

static int
send_packet(int * sock, RecoveryPacket * packet )
{
	int count = 0;
	ReplicateServerInfo * host = NULL;

	host = PGR_get_replicate_server_info();
	if (host == (ReplicateServerInfo*)NULL)
	{
		return STATUS_ERROR;
	}
	count = 0;
	while (send_recovery_packet(*sock,packet) != STATUS_OK)
	{
		if (count < MAX_RETRY_TIMES )
		{
			count ++;
			continue;
		}
		count = 0;
		close(*sock);
		PGR_Set_Replication_Server_Status(host,DATA_ERR);
		host = PGR_get_replicate_server_info();
		if (host == (ReplicateServerInfo*)NULL)
		{
			return STATUS_ERROR;
		}
		PGR_Set_Replication_Server_Status(host,DATA_USE);
		PGR_Create_Socket_Connect(sock, host->hostName , host->recoveryPortNumber);
	}
	return STATUS_OK;
}

static void
master_loop(int fd)
{
	int count;
	int sock;
	int status = STATUS_OK;
	RecoveryPacket packet;
	int r_size = 0;
	bool loop_end = false;

	count = 0;
	while ((status = PGR_Create_Acception(fd,&sock,"",RecoveryPortNumber)) != STATUS_OK)
	{
		PGR_Close_Sock(&sock);
		sock = -1;
		if ( count > MAX_RETRY_TIMES)
		{
			return;
		}
		count ++;
	}
	for(;;)
	{
		int	rtn;
		fd_set	  rmask;
		struct timeval timeout;

		timeout.tv_sec = PGR_RECV_TIMEOUT;
		timeout.tv_usec = 0;

		/*
		 * Wait for something to happen.
		 */
		FD_ZERO(&rmask);
		FD_SET(sock,&rmask);
		rtn = select(sock+1, &rmask, (fd_set *)NULL, (fd_set *)NULL, &timeout);
		if (rtn && FD_ISSET(sock, &rmask))
		{
			r_size = read_packet(sock,&packet);
			if (r_size == 0)
			{
				continue;
			}
			else if (r_size < 0)
			{
			    loop_end=true;
			    break;
		        }
		}
		else
		{
			continue;
		}
		switch (ntohs(packet.packet_no))
		{
			case RECOVERY_PGDATA_REQ :
				/*
				 * PGDATA information request
				 */
				/*
				 * get master server information
				 */
				memset(&packet,0,sizeof(packet));
				set_recovery_packet(&packet, RECOVERY_PGDATA_ANS) ;
				status = send_packet(&sock,&packet);
				PGR_Set_Cluster_Status(STATUS_RECOVERY);
				break;
			case RECOVERY_FSYNC_REQ : 
				/*
				 * get master server information
				 */
				memset(&packet,0,sizeof(packet));
				set_recovery_packet(&packet, RECOVERY_FSYNC_ANS );
				status = send_packet(&sock,&packet);
				PGR_Set_Cluster_Status(STATUS_RECOVERY);
				loop_end = true;
				break;
			case RECOVERY_ERROR_TARGET_ONLY:	
				memset(&packet,0,sizeof(packet));
				set_recovery_packet(&packet, RECOVERY_ERROR_ANS );
				status = send_packet(&sock,&packet);
				PGR_Set_Cluster_Status(STATUS_REPLICATE);
				break;
			case RECOVERY_ERROR_CONNECTION:
				memset(&packet,0,sizeof(packet));
				set_recovery_packet(&packet, RECOVERY_ERROR_ANS );
				status = send_packet(&sock,&packet);
				PGR_Set_Cluster_Status(STATUS_REPLICATE);
				/**
				 * kill broken cluster db.
				 * FIXME: missing MyProcPid here. It must be postmaster's pid.
				 * but here's a bug MyProcPid doesn't initialized properly , so MyProcPid = postmaster's pid.				
				 * To fix this, define variable to set posmaster's pid.
				 */
				kill(MyProcPid,SIGQUIT);
				loop_end = true;
				break;
			case RECOVERY_ERROR_ANS:
			  /* TODO: recovery failed. close this postmaster */
			        loop_end = true;
			        break;
			case RECOVERY_FINISH:
				PGR_Set_Cluster_Status(STATUS_REPLICATE);
				loop_end = true;
				break;
             		default:
		                loop_end = true;
				break;
		}
		if (loop_end)
		{
			break;
		}
	}
	PGR_Close_Sock(&sock);
}

int
PGR_Master_Main(void)
{
	int status;
	int fd = -1;
	int rtn;
	int pid;

	if ((pid = fork()) != 0 )
	{
		return pid;
	}
	
	memset(Local_Host_Name,0,sizeof(Local_Host_Name));
	gethostname(Local_Host_Name,sizeof(Local_Host_Name));
	pqsignal(SIGHUP, authdie);
	pqsignal(SIGTERM, authdie);
	pqsignal(SIGINT, authdie);
	pqsignal(SIGQUIT, authdie);
	pqsignal(SIGALRM, authdie);
	PG_SETMASK(&UnBlockSig);

	status = STATUS_ERROR;
	status = PGR_Create_Socket_Bind(&fd, "", RecoveryPortNumber);

	if (status != STATUS_OK)
	{
		return pid;
	}
	for (;;)
	{
		fd_set	  rmask;
		struct timeval timeout;

		timeout.tv_sec = 60;
		timeout.tv_usec = 0;

		/*
		 * Wait for something to happen.
		 */
		FD_ZERO(&rmask);
		FD_SET(fd,&rmask);
		rtn = select(fd+1, &rmask, (fd_set *)NULL, (fd_set *)NULL, &timeout);
		if (rtn && FD_ISSET(fd, &rmask))
		{
			master_loop(fd);
		}
	}
	return pid;
}

static int
start_recovery_send(int * sock, ReplicateServerInfo * host)
{
	int status;
	RecoveryPacket packet;
	status = PGR_Create_Socket_Connect(sock, host->hostName, host->recoveryPortNumber);
	if (status != STATUS_OK)
	{
		if (Debug_pretty_print)
		{
			elog(DEBUG1,"connection error to replication server");
		}
		return STATUS_ERROR;
	}

	memset(&packet,0,sizeof(packet));
	set_recovery_packet(&packet, RECOVERY_PREPARE_REQ );
	status = send_packet(sock,&packet);

	return status;
}

static int
stop_recovery_send(int * sock, ReplicateServerInfo * host)
{
	int status;
	RecoveryPacket packet;

	memset(&packet,0,sizeof(packet));
	set_recovery_packet(&packet, RECOVERY_ERROR_ANS );
	status = send_packet(sock,&packet);
	return status;
}

static int
direct_send_packet(int packet_no)
{

	int status;
	int fd = -1;
	ReplicateServerInfo * host;
	RecoveryPacket packet;

	host = PGR_get_replicate_server_info();
	if (host == NULL)
	{
		return STATUS_ERROR;
	}
	status = PGR_Create_Socket_Connect(&fd, host->hostName, host->recoveryPortNumber);
	if (status != STATUS_OK)
	{
		PGR_Set_Replication_Server_Status(host,DATA_ERR);
		return STATUS_ERROR;
	}

	memset(&packet,0,sizeof(packet));
	set_recovery_packet(&packet, packet_no );
	status = send_packet(&fd,&packet);

	close(fd);

	return status;
}

int
PGR_recovery_error_send(void)
{
	return direct_send_packet(RECOVERY_ERROR_ANS);
}

int
PGR_recovery_finish_send(void)
{
	return direct_send_packet(RECOVERY_FINISH);
}

static int
rsync_pg_data(char * src, char * dest)
{
	int status;
	char *args[12];
	int pid, i = 0;

	args[i++] = "rsync";
	args[i++] = "-a";
	args[i++] = "-r";
	if (RsyncCompress)
		args[i++] = "-z";
	args[i++] = "--delete";
	args[i++] = "-e";
	args[i++] = RsyncOption;
	args[i++] = src;
	args[i++] = dest;
	args[i++] = NULL;

	pid = fork();
	if (pid == 0)
	{
		status = execv(RsyncPath,args);
	}
	else
	{
		for (;;)
		{
			int result;
			result = wait(&status);
			if (result < 0)
			{
				if (errno == EINTR)
					continue;
				return STATUS_ERROR;
			}

			if (WIFEXITED(status) == 0 || WEXITSTATUS(status) != 0)
				return STATUS_ERROR;
			else
				break;
		}
	}
	return STATUS_OK;
}

static int
remove_dir(char * dir_name)
{
	DIR * dp = NULL;
	struct dirent *dirp = NULL;
	char fname[256];
	int status = 0;

	if ((dp = opendir(dir_name)) == NULL)
	{
		return STATUS_ERROR;
	}
	while ((dirp = readdir(dp)) != NULL)
	{
		if ((!strcmp(dirp->d_name,".")) ||
			(!strcmp(dirp->d_name,"..")))
		{
			continue;
		}
		sprintf(fname,"%s/%s",dir_name,dirp->d_name);
		status = remove(fname);
		if (status < 0)
		{
			remove_dir(fname);
		}
	}
	closedir(dp);
	if (remove(dir_name) < 0)
	{
		return STATUS_ERROR;
	}
	return STATUS_OK;
}

static int
clear_bkup_dir(char * dir_name)
{
	char bkp_dir[256];
	pid_t pid = getpid();

	sprintf(bkp_dir,"%s_%d",dir_name,pid);
	return (remove_dir(bkp_dir));
}

static int
bkup_dir(char * dir_name)
{
	int status;
	char org_dir[256];
	char bkp_dir[256];
	pid_t pid = getpid();

	sprintf(org_dir,"%s",dir_name);
	sprintf(bkp_dir,"%s_%d",dir_name,pid);
	status = rename(org_dir,bkp_dir);
	if (status < 0)
	{
		return STATUS_ERROR;
	}
	return STATUS_OK;
}

static int
restore_dir(char * dir_name)
{
	int status;
	char org_dir[256];
	char bkp_dir[256];
	pid_t pid = getpid();

	sprintf(org_dir,"%s",dir_name);
	sprintf(bkp_dir,"%s_%d",dir_name,pid);
	status = rename(bkp_dir,org_dir);
	if (status < 0)
	{
		remove_dir(org_dir);
		status = rename(bkp_dir,org_dir);
		if (status < 0)
		{
			return STATUS_ERROR;
		}
	}
	return STATUS_OK;
}

static int
rsync_global_dir(char * src, char * dest)
{
	int status;
	char control_file[256];
	char org_dir[256];
	char src_dir[256];
	struct stat fstat;

	sprintf(org_dir,"%s/global",dest);
	sprintf(control_file,"%s/global/pg_control",dest);
	if (bkup_dir(org_dir) != STATUS_OK)
	{
		return STATUS_ERROR;
	}
	sprintf(src_dir,"%s/global",src);
	status = rsync_pg_data(src_dir, dest);
	if (status != STATUS_OK )
	{
		restore_dir(org_dir);
		return STATUS_ERROR;
	}
	/* check pg_control file */
	sleep(3);
	status = stat(control_file,&fstat);
	if (status < 0)
	{
		restore_dir(org_dir);
		return STATUS_ERROR;
	}
	clear_bkup_dir(org_dir);
	return STATUS_OK;
}

static int
first_recovery(char * src, char * dest, char * dir)
{
	int status = STATUS_OK;
	char src_dir[256];
	char dest_dir[256];

	memset(src_dir,0,sizeof(src_dir));
	memset(dest_dir,0,sizeof(dest_dir));
	sprintf(src_dir,"%s/%s",src,dir);
	sprintf(dest_dir,"%s/%s",dest,dir);
	status = bkup_dir(dest_dir);
	if (status < 0)
	{
		return STATUS_ERROR;
	}
	status = rsync_pg_data(src_dir, dest);
	if (status != STATUS_OK )
	{
		restore_dir(dest_dir);
		return STATUS_ERROR;
	}
	return STATUS_OK;
}

static int
second_recovery(char * src, char * dest, char * dir)
{
	int status = STATUS_OK;
	char src_dir[256];
	char dest_dir[256];

	memset(src_dir,0,sizeof(src_dir));
	memset(dest_dir,0,sizeof(dest_dir));
	sprintf(src_dir,"%s/%s",src,dir);
	sprintf(dest_dir,"%s/%s",dest,dir);

	status = rsync_pg_data(src_dir, dest);
	if (status != STATUS_OK )
	{
		restore_dir(dest_dir);
		return STATUS_ERROR;
	}
	clear_bkup_dir(dest_dir);

	return STATUS_OK;
}

static int
recovery_rsync(char * src , char * dest, int stage)
{
	if ((src== NULL) || ( dest == NULL))
	{
		return STATUS_ERROR;
	}

	/* recovery step of "global" directory */
	fprintf(stderr,"%s recovery step of [global] directory...",
			((stage == 1)?"1st":"2nd"));
	if (rsync_global_dir(src, dest) != STATUS_OK)
	{
		fprintf(stderr,"NG\n");
		return STATUS_ERROR;
	}
	fprintf(stderr,"OK\n");

	if (stage == PGR_1ST_RECOVERY)
	{
		/* 1st recovery step of "base" directory */
		fprintf(stderr,"1st recovery step of [base] directory...");
		if (first_recovery(src,dest,"base") != STATUS_OK)
		{
			fprintf(stderr,"NG\n");
			return STATUS_ERROR;
		}
		fprintf(stderr,"OK\n");

		fprintf(stderr,"1st recovery step of [pg_clog] directory...");
		/* 1st recovery step of "pg_clog" directory */
		if (first_recovery(src,dest,"pg_clog") != STATUS_OK)
		{
			fprintf(stderr,"NG\n");
			return STATUS_ERROR;
		}
		fprintf(stderr,"OK\n");

		/* 1st recovery step of "pg_xlog" directory */
		fprintf(stderr,"1st recovery step of [pg_xlog] directory...");
		if (first_recovery(src,dest,"pg_xlog") != STATUS_OK)
		{
			fprintf(stderr,"NG\n");
			return STATUS_ERROR;
		}
		fprintf(stderr,"OK\n");
	}
	else
	{
		/* 2nd recovery step of "base" directory */
		fprintf(stderr,"2nd recovery step of [base] directory...");
		if (second_recovery(src,dest,"base") != STATUS_OK)
		{
			fprintf(stderr,"NG\n");
			return STATUS_ERROR;
		}
		fprintf(stderr,"OK\n");

		/* 2nd recovery step of "pg_clog" directory */
		fprintf(stderr,"2nd recovery step of [pg_clog] directory...");
		if (second_recovery(src,dest,"pg_clog") != STATUS_OK)
		{
			fprintf(stderr,"NG\n");
			return STATUS_ERROR;
		}
		fprintf(stderr,"OK\n");

		/* 2nd recovery step of "pg_xlog" directory */
		fprintf(stderr,"2nd recovery step of [pg_xlog] directory...");
		if (second_recovery(src,dest,"pg_xlog") != STATUS_OK)
		{
			fprintf(stderr,"NG\n");
			return STATUS_ERROR;
		}
		fprintf(stderr,"OK\n");
	}

	return STATUS_OK;
}

static int
recovery_loop(int fd)
{

	int status = STATUS_OK;
	RecoveryPacket packet;
	int r_size = 0;
	int rtn = RECOVERY_LOOP_END;
	char src[256];

	r_size = read_packet(fd,&packet);
	if (r_size <= 0)
	{
		rtn = RECOVERY_LOOP_FAIL;
	}
	switch (ntohs(packet.packet_no))
	{
		case RECOVERY_PREPARE_ANS :
			/*
			 * get master information
			 */
			/*
			 * rsync master data before recovery
			 */
			if (Debug_pretty_print)
			{
				elog(DEBUG1,"local host : %s  master:%s",Local_Host_Name,packet.hostName);
			}
			if (!memcmp(Local_Host_Name,packet.hostName,strlen(packet.hostName)))
			{
				strcpy(src,packet.pg_data);
			}
			else
			{
				sprintf(src,"%s:%s",packet.hostName,packet.pg_data);
			}
			status = recovery_rsync(src,DataDir,PGR_1ST_RECOVERY);
			if (status != STATUS_OK)
			{
				if (Debug_pretty_print)
				{
					elog(DEBUG1,"1st rsync error");
				}
				rtn = RECOVERY_LOOP_FAIL;
				break;
			}
			/*
			 * send recovery start request
			 */
			PGRset_recovery_packet_no(&packet, RECOVERY_START_REQ );
			status = send_packet(&fd,&packet);
			if (status != STATUS_OK)
			{
				rtn = RECOVERY_LOOP_FAIL;
				break;
			}
			rtn = RECOVERY_LOOP_CONTINUE;
			break;
		case RECOVERY_START_ANS : 
			/*
			 * rsync master data before recovery
			 */
			if (!memcmp(Local_Host_Name,packet.hostName,strlen(packet.hostName)))
			{
				strcpy(src,packet.pg_data);
			}
			else
			{
				sprintf(src,"%s:%s",packet.hostName,packet.pg_data);
			}
			status = recovery_rsync(src,DataDir,PGR_2ND_RECOVERY);
			if (status != STATUS_OK)
			{
				if (Debug_pretty_print)
				{
					elog(DEBUG1,"2nd rsync error");
				}
				rtn = RECOVERY_LOOP_FAIL;
				break;
			}
			rtn = RECOVERY_LOOP_END;
			break;
		case RECOVERY_ERROR_OCCUPIED:
		case RECOVERY_ERROR_CONNECTION:
	default:
			rtn = RECOVERY_LOOP_FAIL;
			break;
	}

	return rtn;
}

int
PGR_Recovery_Main(void)
{
	int status;
	int fd = -1;
	int rtn;
	ReplicateServerInfo * host;

	memset(Local_Host_Name,0,sizeof(Local_Host_Name));
	gethostname(Local_Host_Name,sizeof(Local_Host_Name));

	status = STATUS_ERROR;

Retry_Start_Recovery:
	host = PGR_get_replicate_server_info();
	if (host == NULL)
	{
		if (Debug_pretty_print)
		{
			elog(DEBUG1,"not found replication server");
		}
		return STATUS_ERROR;
	}

	status = start_recovery_send(&fd,host);
	if (status != STATUS_OK)
	{
		PGR_Set_Replication_Server_Status(host,DATA_ERR);
		close(fd);
		if (Debug_pretty_print)
		{
			elog(DEBUG1,"start recovery packet send error");
		}
		goto Retry_Start_Recovery;
	}

	for (;;)
	{
		fd_set	  rmask;
		struct timeval timeout;

		timeout.tv_sec = RECOVERY_TIMEOUT;
		timeout.tv_usec = 0;

		/*
		 * Wait for something to happen.
		 */
		FD_ZERO(&rmask);
		FD_SET(fd,&rmask);
		rtn = select(fd+1, &rmask, (fd_set *)NULL, (fd_set *)NULL, &timeout);
		if (rtn && FD_ISSET(fd, &rmask))
		{
			status = recovery_loop(fd);
			if (status == RECOVERY_LOOP_CONTINUE)
			{
				continue;
			}
			else if (status == RECOVERY_LOOP_END)
			{
				close(fd);
				break;
			}
			else if (status == RECOVERY_LOOP_FAIL)
			{
				status = stop_recovery_send(&fd,host);
				if (status != STATUS_OK)
				{
					close(fd);
					return STATUS_ERROR;
				}
				close(fd);
				return STATUS_ERROR;
			}
			else 
			  {
			    close(fd);
			    return STATUS_ERROR;
			  }
		}
	}
	return STATUS_OK;
}

static void
show_recovery_packet(RecoveryPacket * packet)
{

	if (Debug_pretty_print)
	{
		elog(DEBUG1,"no = %d",ntohs(packet->packet_no));
		elog(DEBUG1,"max_connect = %d",ntohs(packet->max_connect));
		elog(DEBUG1,"port = %d",ntohs(packet->port));
		elog(DEBUG1,"recoveryPort = %d",ntohs(packet->recoveryPort));
		if (packet->hostName != NULL)
			elog(DEBUG1,"hostName = %s",packet->hostName);
		if (packet->pg_data != NULL)
			elog(DEBUG1,"pg_data = %s",packet->pg_data);
	}
}

static void
set_recovery_packet(RecoveryPacket * packet, int packet_no)
{
	struct passwd * pw = NULL;

	if (packet == NULL)
	{
		return;
	}
	PGRset_recovery_packet_no(packet, packet_no );
	packet->max_connect = htons(MaxBackends);
	packet->port = htons(PostPortNumber);
	packet->recoveryPort = htons(RecoveryPortNumber);
	gethostname(packet->hostName,sizeof(packet->hostName));
	memcpy(packet->pg_data,DataDir,sizeof(packet->pg_data));
	memset(packet->userName,0,sizeof(packet->userName));
	if ((pw = getpwuid(geteuid())) != NULL)
	{
		strncpy(packet->userName,pw->pw_name,sizeof(packet->userName));
	}
	else
	{
		cuserid(packet->userName);
	}
}

#endif /* USE_REPLICATION */
