src/backend/replication/logical/reorderbuffer.c
void
ReorderBufferQueueMessage(ReorderBuffer *rb, TransactionId xid,
						  Snapshot snapshot, XLogRecPtr lsn,
						  bool transactional, const char *prefix,
						  Size message_size, const char *message)
{
	if (transactional)
	{
		MemoryContext oldcontext;
		ReorderBufferChange *change;

		Assert(xid != InvalidTransactionId);

		oldcontext = MemoryContextSwitchTo(rb->context);

		change = ReorderBufferGetChange(rb);
		change->action = REORDER_BUFFER_CHANGE_MESSAGE;
		change->data.msg.prefix = pstrdup(prefix);
		change->data.msg.message_size = message_size;
		change->data.msg.message = palloc(message_size);
		memcpy(change->data.msg.message, message, message_size);

		ReorderBufferQueueChange(rb, xid, lsn, change);

		MemoryContextSwitchTo(oldcontext);
	}
	else
	{
		ReorderBufferTXN *txn = NULL;
		volatile Snapshot snapshot_now = snapshot;

		if (xid != InvalidTransactionId)
			txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true);

		/* setup snapshot to allow catalog access */
		SetupHistoricSnapshot(snapshot_now, NULL);
		PG_TRY();
		{
			rb->message(rb, txn, lsn, false, prefix, message_size, message);

			TeardownHistoricSnapshot(false);
		}
		PG_CATCH();
		{
			TeardownHistoricSnapshot(true);
			PG_RE_THROW();
		}
		PG_END_TRY();
	}
}
/*
 * Queue message into a transaction so it can be processed upon commit.
 */
Queue message into a transaction so it can be processed upon commit.
Queue message into a transaction so it can be processed upon commit .
void
ReorderBufferQueueMessage(ReorderBuffer *rb, TransactionId xid,
						  Snapshot snapshot, XLogRecPtr lsn,
						  bool transactional, const char *prefix,
						  Size message_size, const char *message)
{
	if (transactional)
	{
		MemoryContext oldcontext;
		ReorderBufferChange *change;
		Assert(xid != InvalidTransactionId);
		oldcontext = MemoryContextSwitchTo(rb->context);
		change = ReorderBufferGetChange(rb);
		change->action = REORDER_BUFFER_CHANGE_MESSAGE;
		change->data.msg.prefix = pstrdup(prefix);
		change->data.msg.message_size = message_size;
		change->data.msg.message = palloc(message_size);
		memcpy(change->data.msg.message, message, message_size);
		ReorderBufferQueueChange(rb, xid, lsn, change);
		MemoryContextSwitchTo(oldcontext);
	}
	else
	{
		ReorderBufferTXN *txn = NULL;
		volatile Snapshot snapshot_now = snapshot;
		if (xid != InvalidTransactionId)
			txn = ReorderBufferTXNByXid(rb, xid, true, NULL, lsn, true);
		SetupHistoricSnapshot(snapshot_now, NULL);
		PG_TRY();
		{
			rb->message(rb, txn, lsn, false, prefix, message_size, message);
			TeardownHistoricSnapshot(false);
		}
		PG_CATCH();
		{
			TeardownHistoricSnapshot(true);
			PG_RE_THROW();
		}
		PG_END_TRY();
	}
}
void ReorderBufferQueueMessage ( ReorderBuffer * rb , TransactionId xid , Snapshot snapshot , XLogRecPtr lsn , bool transactional , const char * prefix , Size message_size , const char * message ) { if ( transactional ) { MemoryContext oldcontext ; ReorderBufferChange * change ; Assert ( xid != InvalidTransactionId ) ; oldcontext = MemoryContextSwitchTo ( rb -> context ) ; change = ReorderBufferGetChange ( rb ) ; change -> action = REORDER_BUFFER_CHANGE_MESSAGE ; change -> data . msg . prefix = pstrdup ( prefix ) ; change -> data . msg . message_size = message_size ; change -> data . msg . message = palloc ( message_size ) ; memcpy ( change -> data . msg . message , message , message_size ) ; ReorderBufferQueueChange ( rb , xid , lsn , change ) ; MemoryContextSwitchTo ( oldcontext ) ; } else { ReorderBufferTXN * txn = NULL ; volatile Snapshot snapshot_now = snapshot ; if ( xid != InvalidTransactionId ) txn = ReorderBufferTXNByXid ( rb , xid , true , NULL , lsn , true ) ; SetupHistoricSnapshot ( snapshot_now , NULL ) ; PG_TRY ( ) ; { rb -> message ( rb , txn , lsn , false , prefix , message_size , message ) ; TeardownHistoricSnapshot ( false ) ; } PG_CATCH ( ) ; { TeardownHistoricSnapshot ( true ) ; PG_RE_THROW ( ) ; } PG_END_TRY ( ) ; } }
Queue message into a transaction so it can be processed upon commit.
Queue message into a transaction so it can be processed upon commit .
/* setup snapshot to allow catalog access */