CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
{
	const char *slot_name;
	const char *snapshot_name = NULL;
	char		xpos[MAXFNAMELEN];
	StringInfoData buf;

	Assert(!MyReplicationSlot);

	/* setup state for XLogReadPage */
	sendTimeLineIsHistoric = false;
	sendTimeLine = ThisTimeLineID;

	if (cmd->kind == REPLICATION_KIND_PHYSICAL)
	{
		ReplicationSlotCreate(cmd->slotname, false, RS_PERSISTENT);
	}
	else
	{
		CheckLogicalDecodingRequirements();
		/*
		 * Initially create the slot as ephemeral - that allows us to nicely
		 * handle errors during initialization because it'll get dropped if
		 * this transaction fails. We'll make it persistent at the end.
		 */
		ReplicationSlotCreate(cmd->slotname, true, RS_EPHEMERAL);
	}

	initStringInfo(&output_message);

	slot_name = NameStr(MyReplicationSlot->data.name);

	if (cmd->kind == REPLICATION_KIND_LOGICAL)
	{
		LogicalDecodingContext *ctx;

		ctx = CreateInitDecodingContext(
										cmd->plugin, NIL,
										logical_read_xlog_page,
										WalSndPrepareWrite, WalSndWriteData);

		/*
		 * Signal that we don't need the timeout mechanism. We're just
		 * creating the replication slot and don't yet accept feedback
		 * messages or send keepalives. As we possibly need to wait for
		 * further WAL the walsender would otherwise possibly be killed too
		 * soon.
		 */
		last_reply_timestamp = 0;

		/* build initial snapshot, might take a while */
		DecodingContextFindStartpoint(ctx);

		/*
		 * Export a plain (not of the snapbuild.c type) snapshot to the user
		 * that can be imported into another session.
		 */
		snapshot_name = SnapBuildExportSnapshot(ctx->snapshot_builder);

		/* don't need the decoding context anymore */
		FreeDecodingContext(ctx);

		ReplicationSlotPersist();
	}

	slot_name = NameStr(MyReplicationSlot->data.name);
	snprintf(xpos, sizeof(xpos), "%X/%X",
			 (uint32) (MyReplicationSlot->data.confirmed_flush >> 32),
			 (uint32) MyReplicationSlot->data.confirmed_flush);

	pq_beginmessage(&buf, 'T');
	pq_sendint(&buf, 4, 2);		/* 4 fields */

	/* first field: slot name */
	pq_sendstring(&buf, "slot_name");	/* col name */
	pq_sendint(&buf, 0, 4);		/* table oid */
	pq_sendint(&buf, 0, 2);		/* attnum */
	pq_sendint(&buf, TEXTOID, 4);		/* type oid */
	pq_sendint(&buf, -1, 2);	/* typlen */
	pq_sendint(&buf, 0, 4);		/* typmod */
	pq_sendint(&buf, 0, 2);		/* format code */

	/* second field: LSN at which we became consistent */
	pq_sendstring(&buf, "consistent_point");	/* col name */
	pq_sendint(&buf, 0, 4);		/* table oid */
	pq_sendint(&buf, 0, 2);		/* attnum */
	pq_sendint(&buf, TEXTOID, 4);		/* type oid */
	pq_sendint(&buf, -1, 2);	/* typlen */
	pq_sendint(&buf, 0, 4);		/* typmod */
	pq_sendint(&buf, 0, 2);		/* format code */

	/* third field: exported snapshot's name */
	pq_sendstring(&buf, "snapshot_name");		/* col name */
	pq_sendint(&buf, 0, 4);		/* table oid */
	pq_sendint(&buf, 0, 2);		/* attnum */
	pq_sendint(&buf, TEXTOID, 4);		/* type oid */
	pq_sendint(&buf, -1, 2);	/* typlen */
	pq_sendint(&buf, 0, 4);		/* typmod */
	pq_sendint(&buf, 0, 2);		/* format code */

	/* fourth field: output plugin */
	pq_sendstring(&buf, "output_plugin");		/* col name */
	pq_sendint(&buf, 0, 4);		/* table oid */
	pq_sendint(&buf, 0, 2);		/* attnum */
	pq_sendint(&buf, TEXTOID, 4);		/* type oid */
	pq_sendint(&buf, -1, 2);	/* typlen */
	pq_sendint(&buf, 0, 4);		/* typmod */
	pq_sendint(&buf, 0, 2);		/* format code */

	pq_endmessage(&buf);

	/* Send a DataRow message */
	pq_beginmessage(&buf, 'D');
	pq_sendint(&buf, 4, 2);		/* # of columns */

	/* slot_name */
	pq_sendint(&buf, strlen(slot_name), 4);		/* col1 len */
	pq_sendbytes(&buf, slot_name, strlen(slot_name));

	/* consistent wal location */
	pq_sendint(&buf, strlen(xpos), 4);	/* col2 len */
	pq_sendbytes(&buf, xpos, strlen(xpos));

	/* snapshot name */
	if (snapshot_name != NULL)
	{
		pq_sendint(&buf, strlen(snapshot_name), 4);		/* col3 len */
		pq_sendbytes(&buf, snapshot_name, strlen(snapshot_name));
	}
	else
		pq_sendint(&buf, -1, 4);	/* col3 len, NULL */

	/* plugin */
	if (cmd->plugin != NULL)
	{
		pq_sendint(&buf, strlen(cmd->plugin), 4);		/* col4 len */
		pq_sendbytes(&buf, cmd->plugin, strlen(cmd->plugin));
	}
	else
		pq_sendint(&buf, -1, 4);	/* col4 len, NULL */

	pq_endmessage(&buf);

	/*
	 * release active status again, START_REPLICATION will reacquire it
	 */
	ReplicationSlotRelease();
}