diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c
index 5739d2b40f..9f62d4ca6b 100644
--- a/src/backend/commands/async.c
+++ b/src/backend/commands/async.c
@@ -1,3 +1,5 @@
+#include <include.h>
+
/*-------------------------------------------------------------------------
  *
  * async.c
@@ -46,7 +48,7 @@
  *    to. In case there is a match it delivers the notification event to its
  *    frontend.  Non-matching events are simply skipped.
  *
- * 4. The NOTIFY statement (routine Async_Notify) stores the notification in
+ * 4. The NOTIFY statement (routine Async_Notify_My) stores the notification in
  *    a backend-local list which will not be processed until transaction end.
  *
  *    Duplicate notifications from the same transaction are sent out as one
@@ -56,7 +58,7 @@
  *    that has been sent, it can easily add some unique string into the extra
  *    payload parameter.
  *
- *    When the transaction is ready to commit, PreCommit_Notify() adds the
+ *    When the transaction is ready to commit, PreCommit_Notify_My() adds the
  *    pending notifications to the head of the queue. The head pointer of the
  *    queue always points to the next free position and a position is just a
  *    page number and the offset in that page. This is done before marking the
@@ -67,7 +69,7 @@
  *    Once we have put all of the notifications into the queue, we return to
  *    CommitTransaction() which will then do the actual transaction commit.
  *
- *    After commit we are called another time (AtCommit_Notify()). Here we
+ *    After commit we are called another time (AtCommit_Notify_My()). Here we
  *    make the actual updates to the effective listen state (listenChannels).
  *
  *    Finally, after we are out of the transaction altogether, we check if
@@ -171,7 +173,7 @@ typedef struct AsyncQueueEntry
{
  int      length;      /* total allocated length of entry */
  Oid      dboid;      /* sender's database OID */
-  TransactionId xid;      /* sender's XID */
+//  TransactionId xid;      /* sender's XID */
  int32    srcPid;      /* sender's PID */
  char    data[NAMEDATALEN + NOTIFY_PAYLOAD_MAX_LENGTH];
} AsyncQueueEntry;
@@ -414,14 +416,16 @@ typedef struct NotificationHash
static NotificationList *pendingNotifies = NULL;
+static pqsigfunc pg_async_signal_original = NULL;
+
/*
- * Inbound notifications are initially processed by HandleNotifyInterrupt(),
+ * Inbound notifications are initially processed by HandleNotifyInterruptMy(),
  * called from inside a signal handler. That just sets the
  * notifyInterruptPending flag and sets the process
- * latch. ProcessNotifyInterrupt() will then be called whenever it's safe to
+ * latch. ProcessNotifyInterruptMy() will then be called whenever it's safe to
  * actually deal with the interrupt.
  */
-volatile sig_atomic_t notifyInterruptPending = false;
+//volatile sig_atomic_t notifyInterruptPending = false;
/* True if we've registered an on_shmem_exit cleanup */
static bool unlistenExitRegistered = false;
@@ -436,7 +440,7 @@ static bool backendHasSentNotifications = false;
static bool backendTryAdvanceTail = false;
/* GUC parameter */
-bool    Trace_notify = false;
+//bool    Trace_notify = false;
/* local function prototypes */
static int  asyncQueuePageDiff(int p, int q);
@@ -469,6 +473,12 @@ static uint32 notification_hash(const void *key, Size keysize);
static int  notification_match(const void *key1, const void *key2, Size keysize);
static void ClearPendingActionsAndNotifies(void);
+static void pg_async_signal(SIGNAL_ARGS) {
+    HandleNotifyInterruptMy();
+    if (notifyInterruptPending) ProcessNotifyInterruptMy();
+    pg_async_signal_original(postgres_signal_arg);
+}
+
/*
  * Compute the difference between two queue page numbers (i.e., p - q),
  * accounting for wraparound.
@@ -509,11 +519,11 @@ asyncQueuePagePrecedes(int p, int q)
  * Report space needed for our shared memory area
  */
Size
-AsyncShmemSize(void)
+AsyncShmemSizeMy(void)
{
  Size    size;
-  /* This had better match AsyncShmemInit */
+  /* This had better match AsyncShmemInitMy */
  size = mul_size(MaxBackends + 1, sizeof(QueueBackendStatus));
  size = add_size(size, offsetof(AsyncQueueControl, backend));
@@ -526,7 +536,7 @@ AsyncShmemSize(void)
  * Initialize our shared memory area
  */
void
-AsyncShmemInit(void)
+AsyncShmemInitMy(void)
{
  bool    found;
  Size    size;
@@ -585,7 +595,7 @@ AsyncShmemInit(void)
  *    SQL function to send a notification event
  */
Datum
-pg_notify(PG_FUNCTION_ARGS)
+pg_notify_my(PG_FUNCTION_ARGS)
{
  const char *channel;
  const char *payload;
@@ -601,16 +611,16 @@ pg_notify(PG_FUNCTION_ARGS)
    payload = text_to_cstring(PG_GETARG_TEXT_PP(1));
  /* For NOTIFY as a statement, this is checked in ProcessUtility */
-  PreventCommandDuringRecovery("NOTIFY");
+//  PreventCommandDuringRecovery("NOTIFY");
-  Async_Notify(channel, payload);
+  Async_Notify_My(channel, payload);
  PG_RETURN_VOID();
}
/*
- * Async_Notify
+ * Async_Notify_My
  *
  *    This is executed by the SQL notify command.
  *
@@ -619,7 +629,7 @@ pg_notify(PG_FUNCTION_ARGS)
  *    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  */
void
-Async_Notify(const char *channel, const char *payload)
+Async_Notify_My(const char *channel, const char *payload)
{
  int      my_level = GetCurrentTransactionNestLevel();
  size_t    channel_len;
@@ -631,7 +641,7 @@ Async_Notify(const char *channel, const char *payload)
    elog(ERROR, "cannot send notifications from a parallel worker");
  if (Trace_notify)
-    elog(DEBUG1, "Async_Notify(%s)", channel);
+    elog(DEBUG1, "Async_Notify_My(%s)", channel);
  channel_len = channel ? strlen(channel) : 0;
  payload_len = payload ? strlen(payload) : 0;
@@ -679,7 +689,7 @@ Async_Notify(const char *channel, const char *payload)
    /*
    * First notify event in current (sub)xact. Note that we allocate the
    * NotificationList in TopTransactionContext; the nestingLevel might
-    * get changed later by AtSubCommit_Notify.
+    * get changed later by AtSubCommit_Notify_My.
    */
    notifies = (NotificationList *)
      MemoryContextAlloc(TopTransactionContext,
@@ -725,7 +735,7 @@ queue_listen(ListenActionKind action, const char *channel)
  int      my_level = GetCurrentTransactionNestLevel();
  /*
-  * Unlike Async_Notify, we don't try to collapse out duplicates. It would
+  * Unlike Async_Notify_My, we don't try to collapse out duplicates. It would
  * be too complicated to ensure we get the right interactions of
  * conflicting LISTEN/UNLISTEN/UNLISTEN_ALL, and it's unlikely that there
  * would be any performance benefit anyway in sane applications.
@@ -745,7 +755,7 @@ queue_listen(ListenActionKind action, const char *channel)
    /*
    * First action in current sub(xact). Note that we allocate the
    * ActionList in TopTransactionContext; the nestingLevel might get
-    * changed later by AtSubCommit_Notify.
+    * changed later by AtSubCommit_Notify_My.
    */
    actions = (ActionList *)
      MemoryContextAlloc(TopTransactionContext, sizeof(ActionList));
@@ -761,29 +771,29 @@ queue_listen(ListenActionKind action, const char *channel)
}
/*
- * Async_Listen
+ * Async_Listen_My
  *
  *    This is executed by the SQL listen command.
  */
void
-Async_Listen(const char *channel)
+Async_Listen_My(const char *channel)
{
  if (Trace_notify)
-    elog(DEBUG1, "Async_Listen(%s,%d)", channel, MyProcPid);
+    elog(DEBUG1, "Async_Listen_My(%s,%d)", channel, MyProcPid);
  queue_listen(LISTEN_LISTEN, channel);
}
/*
- * Async_Unlisten
+ * Async_Unlisten_My
  *
  *    This is executed by the SQL unlisten command.
  */
void
-Async_Unlisten(const char *channel)
+Async_Unlisten_My(const char *channel)
{
  if (Trace_notify)
-    elog(DEBUG1, "Async_Unlisten(%s,%d)", channel, MyProcPid);
+    elog(DEBUG1, "Async_Unlisten_My(%s,%d)", channel, MyProcPid);
  /* If we couldn't possibly be listening, no need to queue anything */
  if (pendingActions == NULL && !unlistenExitRegistered)
@@ -793,15 +803,15 @@ Async_Unlisten(const char *channel)
}
/*
- * Async_UnlistenAll
+ * Async_UnlistenAll_My
  *
  *    This is invoked by UNLISTEN * command, and also at backend exit.
  */
void
-Async_UnlistenAll(void)
+Async_UnlistenAll_My(void)
{
  if (Trace_notify)
-    elog(DEBUG1, "Async_UnlistenAll(%d)", MyProcPid);
+    elog(DEBUG1, "Async_UnlistenAll_My(%d)", MyProcPid);
  /* If we couldn't possibly be listening, no need to queue anything */
  if (pendingActions == NULL && !unlistenExitRegistered)
@@ -818,7 +828,7 @@ Async_UnlistenAll(void)
  * change within a transaction.
  */
Datum
-pg_listening_channels(PG_FUNCTION_ARGS)
+pg_listening_channels_my(PG_FUNCTION_ARGS)
{
  FuncCallContext *funcctx;
@@ -858,13 +868,13 @@ Async_UnlistenOnExit(int code, Datum arg)
}
/*
- * AtPrepare_Notify
+ * AtPrepare_Notify_My
  *
  *    This is called at the prepare phase of a two-phase
  *    transaction.  Save the state for possible commit later.
  */
void
-AtPrepare_Notify(void)
+AtPrepare_Notify_My(void)
{
  /* It's not allowed to have any pending LISTEN/UNLISTEN/NOTIFY actions */
  if (pendingActions || pendingNotifies)
@@ -874,7 +884,7 @@ AtPrepare_Notify(void)
}
/*
- * PreCommit_Notify
+ * PreCommit_Notify_My
  *
  *    This is called at transaction commit, before actually committing to
  *    clog.
@@ -889,7 +899,7 @@ AtPrepare_Notify(void)
  *    we can still throw error if we run out of queue space.
  */
void
-PreCommit_Notify(void)
+PreCommit_Notify_My(void)
{
  ListCell   *p;
@@ -897,7 +907,7 @@ PreCommit_Notify(void)
    return;          /* no relevant statements in this xact */
  if (Trace_notify)
-    elog(DEBUG1, "PreCommit_Notify");
+    elog(DEBUG1, "PreCommit_Notify_My");
  /* Preflight for any pending listen/unlisten actions */
  if (pendingActions != NULL)
@@ -932,7 +942,7 @@ PreCommit_Notify(void)
    * so cheap if we don't, and we'd prefer not to do that work while
    * holding NotifyQueueLock.
    */
-    (void) GetCurrentTransactionId();
+//    (void) GetCurrentTransactionId();
    /*
    * Serialize writers by acquiring a special lock that we hold till
@@ -951,7 +961,7 @@ PreCommit_Notify(void)
    * used by the flatfiles mechanism.)
    */
    LockSharedObject(DatabaseRelationId, InvalidOid, 0,
-            AccessExclusiveLock);
+            RowExclusiveLock);
    /* Now push the notifications into the queue */
    backendHasSentNotifications = true;
@@ -984,14 +994,14 @@ PreCommit_Notify(void)
}
/*
- * AtCommit_Notify
+ * AtCommit_Notify_My
  *
  *    This is called at transaction commit, after committing to clog.
  *
  *    Update listenChannels and clear transaction-local state.
  */
void
-AtCommit_Notify(void)
+AtCommit_Notify_My(void)
{
  ListCell   *p;
@@ -1003,7 +1013,7 @@ AtCommit_Notify(void)
    return;
  if (Trace_notify)
-    elog(DEBUG1, "AtCommit_Notify");
+    elog(DEBUG1, "AtCommit_Notify_My");
  /* Perform any pending listen/unlisten actions */
  if (pendingActions != NULL)
@@ -1036,7 +1046,7 @@ AtCommit_Notify(void)
}
/*
- * Exec_ListenPreCommit --- subroutine for PreCommit_Notify
+ * Exec_ListenPreCommit --- subroutine for PreCommit_Notify_My
  *
  * This function must make sure we are ready to catch any incoming messages.
  */
@@ -1131,7 +1141,7 @@ Exec_ListenPreCommit(void)
}
/*
- * Exec_ListenCommit --- subroutine for AtCommit_Notify
+ * Exec_ListenCommit --- subroutine for AtCommit_Notify_My
  *
  * Add the channel to the list of channels we are listening on.
  */
@@ -1155,10 +1165,12 @@ Exec_ListenCommit(const char *channel)
  oldcontext = MemoryContextSwitchTo(TopMemoryContext);
  listenChannels = lappend(listenChannels, pstrdup(channel));
  MemoryContextSwitchTo(oldcontext);
+
+  if (!pg_async_signal_original) pg_async_signal_original = pqsignal(SIGUSR1, pg_async_signal);
}
/*
- * Exec_UnlistenCommit --- subroutine for AtCommit_Notify
+ * Exec_UnlistenCommit --- subroutine for AtCommit_Notify_My
  *
  * Remove the specified channel name from listenChannels.
  */
@@ -1186,10 +1198,15 @@ Exec_UnlistenCommit(const char *channel)
  * We do not complain about unlistening something not being listened;
  * should we?
  */
+
+  if (!list_length(listenChannels) && pg_async_signal_original) {
+    pqsignal(SIGUSR1, pg_async_signal_original);
+    pg_async_signal_original = NULL;
+  }
}
/*
- * Exec_UnlistenAllCommit --- subroutine for AtCommit_Notify
+ * Exec_UnlistenAllCommit --- subroutine for AtCommit_Notify_My
  *
  *    Unlisten on all channels for this backend.
  */
@@ -1201,10 +1218,15 @@ Exec_UnlistenAllCommit(void)
  list_free_deep(listenChannels);
  listenChannels = NIL;
+
+  if (pg_async_signal_original) {
+    pqsignal(SIGUSR1, pg_async_signal_original);
+    pg_async_signal_original = NULL;
+  }
}
/*
- * ProcessCompletedNotifies --- send out signals and self-notifies
+ * ProcessCompletedNotifiesMy --- send out signals and self-notifies
  *
  * This is called from postgres.c just before going idle at the completion
  * of a transaction.  If we issued any notifications in the just-completed
@@ -1213,10 +1235,10 @@ Exec_UnlistenAllCommit(void)
  * Also, if we filled enough queue pages with new notifies, try to advance
  * the queue tail pointer.
  *
- * The reason that this is not done in AtCommit_Notify is that there is
+ * The reason that this is not done in AtCommit_Notify_My is that there is
  * a nonzero chance of errors here (for example, encoding conversion errors
  * while trying to format messages to our frontend).  An error during
- * AtCommit_Notify would be a PANIC condition.  The timing is also arranged
+ * AtCommit_Notify_My would be a PANIC condition.  The timing is also arranged
  * to ensure that a transaction's self-notifies are delivered to the frontend
  * before it gets the terminating ReadyForQuery message.
  *
@@ -1227,8 +1249,9 @@ Exec_UnlistenAllCommit(void)
  * NOTE: we are outside of any transaction here.
  */
void
-ProcessCompletedNotifies(void)
+ProcessCompletedNotifiesMy(void)
{
+  bool idle = !IsTransactionOrTransactionBlock();
  MemoryContext caller_context;
  /* Nothing to do if we didn't send any notifications */
@@ -1249,12 +1272,13 @@ ProcessCompletedNotifies(void)
  caller_context = CurrentMemoryContext;
  if (Trace_notify)
-    elog(DEBUG1, "ProcessCompletedNotifies");
+    elog(DEBUG1, "ProcessCompletedNotifiesMy");
  /*
  * We must run asyncQueueReadAllNotifications inside a transaction, else
  * bad things happen if it gets an error.
  */
+  if (idle)
  StartTransactionCommand();
  /* Send signals to other backends */
@@ -1275,6 +1299,7 @@ ProcessCompletedNotifies(void)
    asyncQueueAdvanceTail();
  }
+  if (idle)
  CommitTransactionCommand();
  MemoryContextSwitchTo(caller_context);
@@ -1431,7 +1456,7 @@ asyncQueueNotificationToEntry(Notification *n, AsyncQueueEntry *qe)
  entryLength = QUEUEALIGN(entryLength);
  qe->length = entryLength;
  qe->dboid = MyDatabaseId;
-  qe->xid = GetCurrentTransactionId();
+//  qe->xid = GetCurrentTransactionId();
  qe->srcPid = MyProcPid;
  memcpy(qe->data, n->data, channellen + payloadlen + 2);
}
@@ -1567,7 +1592,7 @@ asyncQueueAddEntries(ListCell *nextNotify)
  * occupied.
  */
Datum
-pg_notification_queue_usage(PG_FUNCTION_ARGS)
+pg_notification_queue_usage_my(PG_FUNCTION_ARGS)
{
  double    usage;
@@ -1749,7 +1774,7 @@ SignalBackends(void)
}
/*
- * AtAbort_Notify
+ * AtAbort_Notify_My
  *
  *  This is called at transaction abort.
  *
@@ -1757,10 +1782,10 @@ SignalBackends(void)
  *  executed if the transaction got committed.
  */
void
-AtAbort_Notify(void)
+AtAbort_Notify_My(void)
{
  /*
-  * If we LISTEN but then roll back the transaction after PreCommit_Notify,
+  * If we LISTEN but then roll back the transaction after PreCommit_Notify_My,
  * we have registered as a listener but have not made any entry in
  * listenChannels.  In that case, deregister again.
  */
@@ -1772,12 +1797,12 @@ AtAbort_Notify(void)
}
/*
- * AtSubCommit_Notify() --- Take care of subtransaction commit.
+ * AtSubCommit_Notify_My() --- Take care of subtransaction commit.
  *
  * Reassign all items in the pending lists to the parent transaction.
  */
void
-AtSubCommit_Notify(void)
+AtSubCommit_Notify_My(void)
{
  int      my_level = GetCurrentTransactionNestLevel();
@@ -1844,10 +1869,10 @@ AtSubCommit_Notify(void)
}
/*
- * AtSubAbort_Notify() --- Take care of subtransaction abort.
+ * AtSubAbort_Notify_My() --- Take care of subtransaction abort.
  */
void
-AtSubAbort_Notify(void)
+AtSubAbort_Notify_My(void)
{
  int      my_level = GetCurrentTransactionNestLevel();
@@ -1882,15 +1907,15 @@ AtSubAbort_Notify(void)
}
/*
- * HandleNotifyInterrupt
+ * HandleNotifyInterruptMy
  *
  *    Signal handler portion of interrupt handling. Let the backend know
  *    that there's a pending notify interrupt. If we're currently reading
  *    from the client, this will interrupt the read and
- *    ProcessClientReadInterrupt() will call ProcessNotifyInterrupt().
+ *    ProcessClientReadInterrupt() will call ProcessNotifyInterruptMy().
  */
void
-HandleNotifyInterrupt(void)
+HandleNotifyInterruptMy(void)
{
  /*
  * Note: this is called by a SIGNAL HANDLER. You must be very wary what
@@ -1905,18 +1930,18 @@ HandleNotifyInterrupt(void)
}
/*
- * ProcessNotifyInterrupt
+ * ProcessNotifyInterruptMy
  *
  *    This is called if we see notifyInterruptPending set, just before
  *    transmitting ReadyForQuery at the end of a frontend command, and
  *    also if a notify signal occurs while reading from the frontend.
- *    HandleNotifyInterrupt() will cause the read to be interrupted
+ *    HandleNotifyInterruptMy() will cause the read to be interrupted
  *    via the process's latch, and this routine will get called.
  *    If we are truly idle (ie, *not* inside a transaction block),
  *    process the incoming notifies.
  */
void
-ProcessNotifyInterrupt(void)
+ProcessNotifyInterruptMy(void)
{
  if (IsTransactionOrTransactionBlock())
    return;          /* not really idle */
@@ -1999,7 +2024,7 @@ asyncQueueReadAllNotifications(void)
  * before we see them.
  *----------
  */
-  snapshot = RegisterSnapshot(GetLatestSnapshot());
+//  snapshot = RegisterSnapshot(GetLatestSnapshot());
  /*
  * It is possible that we fail while trying to send a message to our
@@ -2078,7 +2103,7 @@ asyncQueueReadAllNotifications(void)
  PG_END_TRY();
  /* Done with snapshot */
-  UnregisterSnapshot(snapshot);
+//  UnregisterSnapshot(snapshot);
}
/*
@@ -2126,6 +2151,7 @@ asyncQueueProcessPageEntries(volatile QueuePosition *current,
    /* Ignore messages destined for other databases */
    if (qe->dboid == MyDatabaseId)
    {
+#if 0
      if (XidInMVCCSnapshot(qe->xid, snapshot))
      {
        /*
@@ -2153,6 +2179,7 @@ asyncQueueProcessPageEntries(volatile QueuePosition *current,
      }
      else if (TransactionIdDidCommit(qe->xid))
      {
+#endif
        /* qe->data is the null-terminated channel name */
        char     *channel = qe->data;
@@ -2161,8 +2188,9 @@ asyncQueueProcessPageEntries(volatile QueuePosition *current,
          /* payload follows channel name */
          char     *payload = qe->data + strlen(channel) + 1;
-          NotifyMyFrontEnd(channel, payload, qe->srcPid);
+          NotifyMyFrontEndMy(channel, payload, qe->srcPid);
        }
+#if 0
      }
      else
      {
@@ -2171,6 +2199,7 @@ asyncQueueProcessPageEntries(volatile QueuePosition *current,
        * ignore its notifications.
        */
      }
+#endif
    }
    /* Loop back if we're not at end of page */
@@ -2271,6 +2300,7 @@ static void
ProcessIncomingNotify(void)
{
  /* We *must* reset the flag */
+  bool idle = !IsTransactionOrTransactionBlock();
  notifyInterruptPending = false;
  /* Do nothing else if we aren't actively listening */
@@ -2286,10 +2316,12 @@ ProcessIncomingNotify(void)
  * We must run asyncQueueReadAllNotifications inside a transaction, else
  * bad things happen if it gets an error.
  */
+  if (idle)
  StartTransactionCommand();
  asyncQueueReadAllNotifications();
+  if (idle)
  CommitTransactionCommand();
  /*
@@ -2307,7 +2339,7 @@ ProcessIncomingNotify(void)
  * Send NOTIFY message to my front end.
  */
void
-NotifyMyFrontEnd(const char *channel, const char *payload, int32 srcPid)
+NotifyMyFrontEndMy(const char *channel, const char *payload, int32 srcPid)
{
  if (whereToSendOutput == DestRemote)
  {

-- complain if script is sourced in psql, rather than via CREATE EXTENSION
\echo Use "CREATE EXTENSION pg_async" to load this file. \quit
CREATE FUNCTION pg_listen(channel pg_catalog.text default null) RETURNS pg_catalog.void STRICT AS 'MODULE_PATHNAME', 'pg_async_listen' LANGUAGE C;
CREATE FUNCTION pg_listening_channels() RETURNS setof pg_catalog.text STRICT AS 'MODULE_PATHNAME', 'pg_async_listening_channels' LANGUAGE C;
CREATE FUNCTION pg_notification_queue_usage() RETURNS pg_catalog.float8 STRICT AS 'MODULE_PATHNAME', 'pg_async_notification_queue_usage' LANGUAGE C;
CREATE FUNCTION pg_notify(channel pg_catalog.text default null, payload pg_catalog.text default null) RETURNS pg_catalog.void STRICT AS 'MODULE_PATHNAME', 'pg_async_notify' LANGUAGE C;
CREATE FUNCTION pg_unlisten_all() RETURNS pg_catalog.void STRICT AS 'MODULE_PATHNAME', 'pg_async_unlisten_all' LANGUAGE C;
CREATE FUNCTION pg_unlisten(channel pg_catalog.text default null) RETURNS pg_catalog.void STRICT AS 'MODULE_PATHNAME', 'pg_async_unlisten' LANGUAGE C;

#define EXTENSION(function) Datum (function)(PG_FUNCTION_ARGS); PG_FUNCTION_INFO_V1(function); Datum (function)(PG_FUNCTION_ARGS)
EXTENSION(pg_async_listen) {
    const char *channel = PG_ARGISNULL(0) ? "" : text_to_cstring(PG_GETARG_TEXT_PP(0));
    !XactReadOnly ? Async_Listen(channel) : Async_Listen_My(channel);
    PG_RETURN_VOID();
}
EXTENSION(pg_async_listening_channels) {
    return !XactReadOnly ? pg_listening_channels(fcinfo) : pg_listening_channels_my(fcinfo);
}
EXTENSION(pg_async_notification_queue_usage) {
    return !XactReadOnly ? pg_notification_queue_usage(fcinfo) : pg_notification_queue_usage_my(fcinfo);
}
EXTENSION(pg_async_notify) {
    return !XactReadOnly ? pg_notify(fcinfo) : pg_notify_my(fcinfo);
}
EXTENSION(pg_async_unlisten_all) {
    !XactReadOnly ? Async_UnlistenAll() : Async_UnlistenAll_My();
    PG_RETURN_VOID();
}
EXTENSION(pg_async_unlisten) {
    const char *channel = PG_ARGISNULL(0) ? "" : text_to_cstring(PG_GETARG_TEXT_PP(0));
    !XactReadOnly ? Async_Unlisten(channel) : Async_Unlisten_My(channel);
    PG_RETURN_VOID();
}

static ProcessUtility_hook_type pg_async_ProcessUtility_hook_original = NULL;
static shmem_startup_hook_type pg_async_shmem_startup_hook_original = NULL;
void _PG_init(void); void _PG_init(void) {
    if (!process_shared_preload_libraries_in_progress) return;
    pg_async_ProcessUtility_hook_original = ProcessUtility_hook;
    ProcessUtility_hook = pg_async_ProcessUtility_hook;
    pg_async_shmem_startup_hook_original = shmem_startup_hook;
    shmem_startup_hook = pg_async_shmem_startup_hook;
    RequestAddinShmemSpace(AsyncShmemSizeMy());
    RegisterSubXactCallback(pg_async_SubXactCallback, NULL);
    RegisterXactCallback(pg_async_XactCallback, NULL);
}
void _PG_fini(void); void _PG_fini(void) {
    ProcessUtility_hook = pg_async_ProcessUtility_hook_original;
    shmem_startup_hook = pg_async_shmem_startup_hook_original;
    UnregisterSubXactCallback(pg_async_SubXactCallback, NULL);
    UnregisterXactCallback(pg_async_XactCallback, NULL);
}

static void pg_async_shmem_startup_hook(void) {
    if (pg_async_shmem_startup_hook_original) pg_async_shmem_startup_hook_original();
    LWLockAcquire(AddinShmemInitLock, LW_EXCLUSIVE);
    AsyncShmemInitMy();
    LWLockRelease(AddinShmemInitLock);
}

static void pg_async_XactCallback(XactEvent event, void *arg) {
    if (!XactReadOnly) return;
    switch (event) {
        case XACT_EVENT_ABORT: AtAbort_Notify_My(); break;
        case XACT_EVENT_COMMIT: AtCommit_Notify_My(); ProcessCompletedNotifiesMy(); break;
        case XACT_EVENT_PRE_COMMIT: PreCommit_Notify_My(); break;
        case XACT_EVENT_PREPARE: AtPrepare_Notify_My(); break;
        default: break;
    }
}

static void CheckRestrictedOperation(const char *cmdname) {
    if (InSecurityRestrictedOperation()) ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), errmsg("cannot execute %s within security-restricted operation", cmdname)));
}
static void pg_async_ProcessUtility_hook(PlannedStmt *pstmt, const char *queryString, ProcessUtilityContext context, ParamListInfo params, QueryEnvironment *queryEnv, DestReceiver *dest, QueryCompletion *qc) {
    Node *parsetree = pstmt->utilityStmt;
    if (!XactReadOnly) return pg_async_ProcessUtility_hook_original ? pg_async_ProcessUtility_hook_original(pstmt, queryString, context, params, queryEnv, dest, qc) : standard_ProcessUtility(pstmt, queryString, context, params, queryEnv, dest, qc);
    check_stack_depth();
    switch (nodeTag(parsetree)) {
        case T_ListenStmt: {
            ListenStmt *stmt = (ListenStmt *)parsetree;
            CheckRestrictedOperation("LISTEN");
            Async_Listen_My(stmt->conditionname);
        } break;
        case T_NotifyStmt: {
            NotifyStmt *stmt = (NotifyStmt *)parsetree;
            Async_Notify_My(stmt->conditionname, stmt->payload);
        } break;
        case T_UnlistenStmt: {
            UnlistenStmt *stmt = (UnlistenStmt *)parsetree;
            CheckRestrictedOperation("UNLISTEN");
            stmt->conditionname ? Async_Unlisten_My(stmt->conditionname) : Async_UnlistenAll_My();
        } break;
        default: return pg_async_ProcessUtility_hook_original ? pg_async_ProcessUtility_hook_original(pstmt, queryString, context, params, queryEnv, dest, qc) : standard_ProcessUtility(pstmt, queryString, context, params, queryEnv, dest, qc);
    }
    CommandCounterIncrement();
}