Quelle  message.c

/*-------------------------------------------------------------------------
 *
 * message.c
 *   Generic logical messages.
 *
 * Copyright (c) 2013-2025, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *   src/backend/replication/logical/message.c
 *
 * NOTES
 *
 * Generic logical messages allow XLOG logging of arbitrary binary blobs that
 * get passed to the logical decoding plugin. In normal XLOG processing they
 * are same as NOOP.
 *
 * These messages can be either transactional or non-transactional.
 * Transactional messages are part of current transaction and will be sent to
 * decoding plugin using in a same way as DML operations.
 * Non-transactional messages are sent to the plugin at the time when the
 * logical decoding reads them from XLOG. This also means that transactional
 * messages won't be delivered if the transaction was rolled back but the
 * non-transactional one will always be delivered.
 *
 * Every message carries prefix to avoid conflicts between different decoding
 * plugins. The plugin authors must take extra care to use unique prefix,
 * good options seems to be for example to use the name of the extension.
 *
 * ---------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/xact.h"
#include "access/xloginsert.h"
#include "miscadmin.h"
#include "replication/message.h"

/*
 * Write logical decoding message into XLog.
 */
XLogRecPtr
LogLogicalMessage(const char *prefix, const char *message, size_t size,
      bool transactional, bool flush)
{
 xl_logical_message xlrec;
 XLogRecPtr lsn;

 /*
  * Force xid to be allocated if we're emitting a transactional message.
 */
 if (transactional)
 {
  Assert(IsTransactionState());
  GetCurrentTransactionId();
 }

 xlrec.dbId = MyDatabaseId;
 xlrec.transactional = transactional;
 /* trailing zero is critical; see logicalmsg_desc */
 xlrec.prefix_size = strlen(prefix) + 1;
 xlrec.message_size = size;

 XLogBeginInsert();
 XLogRegisterData(&xlrec, SizeOfLogicalMessage);
 XLogRegisterData(prefix, xlrec.prefix_size);
 XLogRegisterData(message, size);

 /* allow origin filtering */
 XLogSetRecordFlags(XLOG_INCLUDE_ORIGIN);

 lsn = XLogInsert(RM_LOGICALMSG_ID, XLOG_LOGICAL_MESSAGE);

 /*
  * Make sure that the message hits disk before leaving if emitting a
  * non-transactional message when flush is requested.
 */
 if (!transactional && flush)
  XLogFlush(lsn);
 return lsn;
}

/*
 * Redo is basically just noop for logical decoding messages.
 */
void
logicalmsg_redo(XLogReaderState *record)
{
 uint8  info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;

 if (info != XLOG_LOGICAL_MESSAGE)
  elog(PANIC, "logicalmsg_redo: unknown op code %u", info);

 /* This is only interesting for logical decoding, see decode.c. */
}