message.c

Go to the documentation of this file.

/*-------------------------------------------------------------------------
 *
 * 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. */
}