[CF 5630] Allow LISTEN on patterns

This branch was automatically generated by a robot using patches from an
email thread registered at:

https://commitfest.postgresql.org/patch/5630

The branch will be overwritten each time a new patch version is posted to
the thread, and also periodically to check for bitrot caused by changes
on the master branch.

Patch(es): https://www.postgresql.org/message-id/[email protected]
Author(s): Zongliang Quan

Author: Commitfest Bot

doc/src/sgml/ref/listen.sgml

@@ -21,7 +21,7 @@ PostgreSQL documentation
 
  <refsynopsisdiv>
 <synopsis>
-LISTEN <replaceable class="parameter">channel</replaceable>
+LISTEN { <replaceable class="parameter">channel</replaceable> | <replaceable>pattern</replaceable> }
 </synopsis>
  </refsynopsisdiv>
 
@@ -30,17 +30,18 @@ LISTEN <replaceable class="parameter">channel</replaceable>
 
   <para>
    <command>LISTEN</command> registers the current session as a
-   listener on the notification channel named <replaceable
-   class="parameter">channel</replaceable>.
+   listener on the notification channels named <replaceable
+   class="parameter">channel</replaceable> or whose name match
+   the <replaceable class="parameter">pattern</replaceable>.
    If the current session is already registered as a listener for
-   this notification channel, nothing is done.
+   these notification channels, nothing is done.
   </para>
 
   <para>
    Whenever the command <command>NOTIFY <replaceable
    class="parameter">channel</replaceable></command> is invoked, either
    by this session or another one connected to the same database, all
-   the sessions currently listening on that notification channel are
+   the sessions currently listening on those notification channels are
    notified, and each will in turn notify its connected client
    application.
   </para>
@@ -77,6 +78,15 @@ LISTEN <replaceable class="parameter">channel</replaceable>
      </para>
     </listitem>
    </varlistentry>
+
+   <varlistentry>
+    <term><replaceable class="parameter">pattern</replaceable></term>
+    <listitem>
+     <para>
+      Pattern of notification channel names (<xref linkend="functions-like"/> expression).
+     </para>
+    </listitem>
+   </varlistentry>
   </variablelist>
  </refsect1>
 
@@ -130,6 +140,17 @@ LISTEN <replaceable class="parameter">channel</replaceable>
 LISTEN virtual;
 NOTIFY virtual;
 Asynchronous notification "virtual" received from server process with PID 8448.
+</programlisting></para>
+  <para>
+   Configure and execute a listen pattern from
+   <application>psql</application>:
+
+<programlisting>
+LISTEN 'virtual%';
+NOTIFY virtual0;
+Asynchronous notification "virtual0" received from server process with PID 8448.
+NOTIFY virtual1;
+Asynchronous notification "virtual1" received from server process with PID 8448.
 </programlisting></para>
  </refsect1>
 

doc/src/sgml/ref/unlisten.sgml

@@ -21,7 +21,7 @@ PostgreSQL documentation
 
  <refsynopsisdiv>
 <synopsis>
-UNLISTEN { <replaceable class="parameter">channel</replaceable> | * }
+UNLISTEN { <replaceable class="parameter">channel</replaceable> | <replaceable class="parameter">pattern</replaceable> | * }
 </synopsis>
  </refsynopsisdiv>
 
@@ -33,10 +33,11 @@ UNLISTEN { <replaceable class="parameter">channel</replaceable> | * }
    registration for <command>NOTIFY</command> events.
    <command>UNLISTEN</command> cancels any existing registration of
    the current <productname>PostgreSQL</productname> session as a
-   listener on the notification channel named <replaceable
-   class="parameter">channel</replaceable>.  The special wildcard
-   <literal>*</literal> cancels all listener registrations for the
-   current session.
+   listener on the notification channels named <replaceable
+   class="parameter">channel</replaceable> or whose name match
+   the <replaceable class="parameter">pattern</replaceable>.
+   The special wildcard <literal>*</literal> cancels all listener
+   registrations for the current session.
   </para>
 
   <para>
@@ -60,6 +61,15 @@ UNLISTEN { <replaceable class="parameter">channel</replaceable> | * }
     </listitem>
    </varlistentry>
 
+   <varlistentry>
+    <term><replaceable class="parameter">pattern</replaceable></term>
+    <listitem>
+     <para>
+      Pattern of notification channel names (<xref linkend="functions-like"/> expression).
+     </para>
+    </listitem>
+   </varlistentry>
+
    <varlistentry>
     <term><literal>*</literal></term>
     <listitem>

src/backend/commands/async.c

@@ -133,6 +133,7 @@
 #include "access/slru.h"
 #include "access/transam.h"
 #include "access/xact.h"
+#include "catalog/pg_collation.h"
 #include "catalog/pg_database.h"
 #include "commands/async.h"
 #include "common/hashfn.h"
@@ -312,6 +313,12 @@ static SlruCtlData NotifyCtlData;
 
 #define QUEUE_FULL_WARN_INTERVAL	5000	/* warn at most once every 5s */
 
+typedef struct
+{
+	bool	ispatt;
+	char	channel[FLEXIBLE_ARRAY_MEMBER]; /* nul-terminated string */
+} ListenChannel;
+
 /*
  * listenChannels identifies the channels we are actually listening to
  * (ie, have committed a LISTEN on).  It is a simple list of channel names,
@@ -339,6 +346,7 @@ typedef enum
 typedef struct
 {
 	ListenActionKind action;
+	bool			 ispatt;
 	char		channel[FLEXIBLE_ARRAY_MEMBER]; /* nul-terminated string */
 } ListenAction;
 
@@ -430,13 +438,13 @@ int			max_notify_queue_pages = 1048576;
 /* local function prototypes */
 static inline int64 asyncQueuePageDiff(int64 p, int64 q);
 static inline bool asyncQueuePagePrecedes(int64 p, int64 q);
-static void queue_listen(ListenActionKind action, const char *channel);
+static void queue_listen(ListenActionKind action, const bool ispatt, const char *channel);
 static void Async_UnlistenOnExit(int code, Datum arg);
 static void Exec_ListenPreCommit(void);
-static void Exec_ListenCommit(const char *channel);
-static void Exec_UnlistenCommit(const char *channel);
+static void Exec_ListenCommit(const bool ispatt, const char *channel);
+static void Exec_UnlistenCommit(const bool ispatt, const char *channel);
 static void Exec_UnlistenAllCommit(void);
-static bool IsListeningOn(const char *channel);
+static bool IsListeningOn(const bool trymatch, const bool ispatt, const char *channel);
 static void asyncQueueUnregister(void);
 static bool asyncQueueIsFull(void);
 static bool asyncQueueAdvance(volatile QueuePosition *position, int entryLength);
@@ -687,7 +695,7 @@ Async_Notify(const char *channel, const char *payload)
  *		commit.
  */
 static void
-queue_listen(ListenActionKind action, const char *channel)
+queue_listen(ListenActionKind action, const bool ispatt, const char *channel)
 {
 	MemoryContext oldcontext;
 	ListenAction *actrec;
@@ -705,6 +713,7 @@ queue_listen(ListenActionKind action, const char *channel)
 	actrec = (ListenAction *) palloc(offsetof(ListenAction, channel) +
 									 strlen(channel) + 1);
 	actrec->action = action;
+	actrec->ispatt = ispatt;
 	strcpy(actrec->channel, channel);
 
 	if (pendingActions == NULL || my_level > pendingActions->nestingLevel)
@@ -735,12 +744,12 @@ queue_listen(ListenActionKind action, const char *channel)
  *		This is executed by the SQL listen command.
  */
 void
-Async_Listen(const char *channel)
+Async_Listen(const bool ispatt, const char *channel)
 {
 	if (Trace_notify)
 		elog(DEBUG1, "Async_Listen(%s,%d)", channel, MyProcPid);
 
-	queue_listen(LISTEN_LISTEN, channel);
+	queue_listen(LISTEN_LISTEN, ispatt, channel);
 }
 
 /*
@@ -749,7 +758,7 @@ Async_Listen(const char *channel)
  *		This is executed by the SQL unlisten command.
  */
 void
-Async_Unlisten(const char *channel)
+Async_Unlisten(const bool ispatt, const char *channel)
 {
 	if (Trace_notify)
 		elog(DEBUG1, "Async_Unlisten(%s,%d)", channel, MyProcPid);
@@ -758,7 +767,7 @@ Async_Unlisten(const char *channel)
 	if (pendingActions == NULL && !unlistenExitRegistered)
 		return;
 
-	queue_listen(LISTEN_UNLISTEN, channel);
+	queue_listen(LISTEN_UNLISTEN, ispatt, channel);
 }
 
 /*
@@ -776,7 +785,7 @@ Async_UnlistenAll(void)
 	if (pendingActions == NULL && !unlistenExitRegistered)
 		return;
 
-	queue_listen(LISTEN_UNLISTEN_ALL, "");
+	queue_listen(LISTEN_UNLISTEN_ALL, false, "");
 }
 
 /*
@@ -803,10 +812,31 @@ pg_listening_channels(PG_FUNCTION_ARGS)
 
 	if (funcctx->call_cntr < list_length(listenChannels))
 	{
-		char	   *channel = (char *) list_nth(listenChannels,
-												funcctx->call_cntr);
+		ListenChannel *chnl;
+
+		chnl = (ListenChannel *)list_nth(listenChannels, funcctx->call_cntr);
+
+		if (chnl->ispatt)
+		{
+			Size plen;
+			char *result;
+			MemoryContext oldcontext;
+
+			oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
+
+			plen = strlen(chnl->channel);
+			result = (char *)palloc(plen + 3);
+			result[0] = '\'';
+			memcpy(result + 1, chnl->channel, plen);
+			result[plen + 1] = '\'';
+			result[plen + 2] = '\0';
 
-		SRF_RETURN_NEXT(funcctx, CStringGetTextDatum(channel));
+			MemoryContextSwitchTo(oldcontext);
+
+			SRF_RETURN_NEXT(funcctx, CStringGetTextDatum(result));
+		}
+		else
+			SRF_RETURN_NEXT(funcctx, CStringGetTextDatum(chnl->channel));
 	}
 
 	SRF_RETURN_DONE(funcctx);
@@ -989,10 +1019,10 @@ AtCommit_Notify(void)
 			switch (actrec->action)
 			{
 				case LISTEN_LISTEN:
-					Exec_ListenCommit(actrec->channel);
+					Exec_ListenCommit(actrec->ispatt, actrec->channel);
 					break;
 				case LISTEN_UNLISTEN:
-					Exec_UnlistenCommit(actrec->channel);
+					Exec_UnlistenCommit(actrec->ispatt, actrec->channel);
 					break;
 				case LISTEN_UNLISTEN_ALL:
 					Exec_UnlistenAllCommit();
@@ -1133,12 +1163,13 @@ Exec_ListenPreCommit(void)
  * Add the channel to the list of channels we are listening on.
  */
 static void
-Exec_ListenCommit(const char *channel)
+Exec_ListenCommit(const bool ispatt, const char *channel)
 {
-	MemoryContext oldcontext;
+	MemoryContext	oldcontext;
+	ListenChannel  *chnl;
 
 	/* Do nothing if we are already listening on this channel */
-	if (IsListeningOn(channel))
+	if (IsListeningOn(false, ispatt, channel))
 		return;
 
 	/*
@@ -1150,7 +1181,15 @@ Exec_ListenCommit(const char *channel)
 	 * later.
 	 */
 	oldcontext = MemoryContextSwitchTo(TopMemoryContext);
-	listenChannels = lappend(listenChannels, pstrdup(channel));
+
+	chnl = (ListenChannel *) palloc(offsetof(ListenChannel, channel) +
+								strlen(channel) + 1);
+
+	chnl->ispatt = ispatt;
+	strcpy(chnl->channel, channel);
+
+	listenChannels = lappend(listenChannels, chnl);
+
 	MemoryContextSwitchTo(oldcontext);
 }
 
@@ -1160,7 +1199,7 @@ Exec_ListenCommit(const char *channel)
  * Remove the specified channel name from listenChannels.
  */
 static void
-Exec_UnlistenCommit(const char *channel)
+Exec_UnlistenCommit(const bool ispatt, const char *channel)
 {
 	ListCell   *q;
 
@@ -1169,9 +1208,12 @@ Exec_UnlistenCommit(const char *channel)
 
 	foreach(q, listenChannels)
 	{
-		char	   *lchan = (char *) lfirst(q);
+		ListenChannel *lchan = (ListenChannel *) lfirst(q);
+
+		if (lchan->ispatt != ispatt)
+			continue;
 
-		if (strcmp(lchan, channel) == 0)
+		if (strcmp(lchan->channel, channel) == 0)
 		{
 			listenChannels = foreach_delete_current(listenChannels, q);
 			pfree(lchan);
@@ -1209,16 +1251,37 @@ Exec_UnlistenAllCommit(void)
  * fairly short, though.
  */
 static bool
-IsListeningOn(const char *channel)
+IsListeningOn(const bool trymatch, const bool ispatt, const char *channel)
 {
 	ListCell   *p;
 
 	foreach(p, listenChannels)
 	{
-		char	   *lchan = (char *) lfirst(p);
+		ListenChannel *lchan = (ListenChannel *) lfirst(p);
 
-		if (strcmp(lchan, channel) == 0)
-			return true;
+		if (trymatch)
+		{
+			Assert(!ispatt);
+
+			if (lchan->ispatt)
+			{
+				Datum s = PointerGetDatum(cstring_to_text(channel));
+				Datum p = PointerGetDatum(cstring_to_text(lchan->channel));
+
+				if (DatumGetBool(DirectFunctionCall2Coll(textlike, DEFAULT_COLLATION_OID, s, p)))
+					return true;
+			}
+			else if (strcmp(lchan->channel, channel) == 0)
+				return true;
+		}
+		else
+		{
+			if (ispatt == lchan->ispatt)
+			{
+				if (strcmp(lchan->channel, channel) == 0)
+					return true;
+			}
+		}
 	}
 	return false;
 }
@@ -2071,7 +2134,7 @@ asyncQueueProcessPageEntries(volatile QueuePosition *current,
 				/* qe->data is the null-terminated channel name */
 				char	   *channel = qe->data;
 
-				if (IsListeningOn(channel))
+				if (IsListeningOn(true, false, channel))
 				{
 					/* payload follows channel name */
 					char	   *payload = qe->data + strlen(channel) + 1;

src/backend/parser/gram.y

@@ -11062,6 +11062,14 @@ ListenStmt: LISTEN ColId
 				{
 					ListenStmt *n = makeNode(ListenStmt);
 
+					n->conditionname = $2;
+					$$ = (Node *) n;
+				}
+			| LISTEN Sconst
+				{
+					ListenStmt *n = makeNode(ListenStmt);
+
+					n->ispatt = true;
 					n->conditionname = $2;
 					$$ = (Node *) n;
 				}
@@ -11082,6 +11090,14 @@ UnlistenStmt:
 					n->conditionname = NULL;
 					$$ = (Node *) n;
 				}
+			| UNLISTEN Sconst
+				{
+					UnlistenStmt *n = makeNode(UnlistenStmt);
+
+					n->ispatt = true;
+					n->conditionname = $2;
+					$$ = (Node *) n;
+				}
 		;