constraints as <xref linkend="app-pgreceivexlog">, plus those for logical

replication (see <xref linkend="logicaldecoding">).

</para>

<para>

<command>pg_recvlogical</> has no equivalent to the logical decoding

SQL interface's peek and get modes. It sends replay confirmations for

data lazily as it receives it and on clean exit. To examine pending data on

a slot without consuming it, use

<link linkend="functions-replication"><function>pg_logical_slot_peek_changes</></>.

</para>

</refsect1>

<refsect1>

</listitem>

</varlistentry>

<varlistentry>

<term><option>-E <replaceable>lsn</replaceable></option></term>

<term><option>--endpos=<replaceable>lsn</replaceable></option></term>

<listitem>

<para>

In <option>--start</option> mode, automatically stop replication

and exit with normal exit status 0 when receiving reaches the

specified LSN. If specified when not in <option>--start</option>

mode, an error is raised.

</para>

<para>

If there's a record with LSN exactly equal to <replaceable>lsn</>,

the record will be output.

</para>

<para>

The <option>--endpos</option> option is not aware of transaction

boundaries and may truncate output partway through a transaction.

Any partially output transaction will not be consumed and will be

replayed again when the slot is next read from. Individual messages

are never truncated.

</para>

</listitem>

</varlistentry>

<varlistentry>

<term><option>--if-not-exists</option></term>

<listitem>

staticintstandby_message_timeout=10*1000;/* 10 sec = default */

staticintfsync_interval=10*1000;/* 10 sec = default */

staticXLogRecPtrstartpos=InvalidXLogRecPtr;

staticXLogRecPtrendpos=InvalidXLogRecPtr;

staticbooldo_create_slot= false;

staticboolslot_exists_ok= false;

staticbooldo_start_slot= false;

staticvoidusage(void);

staticvoidStreamLogicalLog(void);

staticvoiddisconnect_and_exit(intcode);

staticboolflushAndSendFeedback(PGconn*conn,TimestampTz*now);

staticvoidprepareToTerminate(PGconn*conn,XLogRecPtrendpos,

boolkeepalive,XLogRecPtrlsn);

usage(void)

" time between fsyncs to the output file (default: %d)\n"), (fsync_interval /1000));

printf(_(" --if-not-exists do not error if slot already exists when creating a slot\n"));

printf(_(" -I, --startpos=LSN where in an existing slot should the streaming start\n"));

printf(_(" -E, --endpos=LSN exit after receiving the specified LSN\n"));

printf(_(" -n, --no-loop do not loop on connection lost\n"));

printf(_(" -o, --option=NAME[=VALUE]\n"

intbytes_written;

int64now;

inthdr_len;

XLogRecPtrcur_record_lsn=InvalidXLogRecPtr;

if (copybuf!=NULL)

{

intpos;

boolreplyRequested;

XLogRecPtrwalEnd;

boolendposReached= false;

}

replyRequested=copybuf[pos];

if (replyRequested)

if (endpos!=InvalidXLogRecPtr&&walEnd >=endpos)

{

if (!OutputFsync(now))

gotoerror;

endposReached= true;

}

now=feGetCurrentTimestamp();

if (!sendFeedback(conn,now, true, false))

if (replyRequested||endposReached)

{

if (!flushAndSendFeedback(conn,&now))

gotoerror;

last_status=now;

}

if (endposReached)

{

prepareToTerminate(conn,endpos, true,InvalidXLogRecPtr);

time_to_abort= true;

break;

}

continue;

}

elseif (copybuf[0]!='w')

gotoerror;

}

}

{

XLogRecPtrtemp=fe_recvint64(&copybuf[1]);

cur_record_lsn=fe_recvint64(&copybuf[1]);

output_written_lsn=Max(temp,output_written_lsn);

if (endpos!=InvalidXLogRecPtr&&cur_record_lsn>endpos)

{

if (!flushAndSendFeedback(conn,&now))

gotoerror;

prepareToTerminate(conn,endpos, false,cur_record_lsn);

time_to_abort= true;

break;

}

output_written_lsn=Max(cur_record_lsn,output_written_lsn);

bytes_left=r-hdr_len;

bytes_written=0;

strerror(errno));

gotoerror;

}

if (endpos!=InvalidXLogRecPtr&&cur_record_lsn==endpos)

{

if (!flushAndSendFeedback(conn,&now))

gotoerror;

prepareToTerminate(conn,endpos, false,cur_record_lsn);

time_to_abort= true;

break;

}

}

res=PQgetResult(conn);

if (PQresultStatus(res)!=PGRES_COMMAND_OK)

if (PQresultStatus(res)==PGRES_COPY_OUT)

{

gotoerror;

}

elseif (PQresultStatus(res)!=PGRES_COMMAND_OK)

{

fprintf(stderr,

_("%s: unexpected termination of replication stream: %s"),

{"password",no_argument,NULL,'W'},

{"startpos",required_argument,NULL,'I'},

{"endpos",required_argument,NULL,'E'},

{"option",required_argument,NULL,'o'},

{"plugin",required_argument,NULL,'P'},

{"status-interval",required_argument,NULL,'s'},

}

}

while ((c=getopt_long(argc,argv,"f:F:nvd:h:p:U:wWI:o:P:s:S:",

while ((c=getopt_long(argc,argv,"f:F:nvd:h:p:U:wWI:E:o:P:s:S:",

long_options,&option_index))!=-1)

{

switch (c)

}

startpos= ((uint64)hi) <<32 |lo;

break;

case'E':

if (sscanf(optarg,"%X/%X",&hi,&lo)!=2)

{

fprintf(stderr,

_("%s: could not parse end position \"%s\"\n"),

progname,optarg);

exit(1);

}

endpos= ((uint64)hi) <<32 |lo;

break;

case'o':

{

char*data=pg_strdup(optarg);

exit(1);

}

if (endpos!=InvalidXLogRecPtr&& !do_start_slot)

{

fprintf(stderr,

_("%s: --endpos may only be specified with --start\n"),

progname);

fprintf(stderr,_("Try \"%s --help\" for more information.\n"),

progname);

exit(1);

}

pqsignal(SIGINT,sigint_handler);

pqsignal(SIGHUP,sighup_handler);

if (time_to_abort)

{

disconnect_and_exit(0);

}

}

}

}

flushAndSendFeedback(PGconn*conn,TimestampTz*now)

{

if (!OutputFsync(*now))

return false;

*now=feGetCurrentTimestamp();

if (!sendFeedback(conn,*now, true, false))

return false;

return true;

}

prepareToTerminate(PGconn*conn,XLogRecPtrendpos,boolkeepalive,XLogRecPtrlsn)

{

(void)PQputCopyEnd(conn,NULL);

(void)PQflush(conn);

if (verbose)

{

if (keepalive)

fprintf(stderr,"%s: endpos %X/%X reached by keepalive\n",

progname,

(uint32) (endpos >>32), (uint32)endpos);

fprintf(stderr,"%s: endpos %X/%X reached by record at %X/%X\n",

progname, (uint32) (endpos >>32), (uint32) (endpos),

(uint32) (lsn >>32), (uint32)lsn);

}

}