Re: async notification patch for dblink - Mailing list pgsql-hackers

-----BEGIN PGP SIGNED MESSAGE-----
Hash: SHA256

Alvaro Herrera wrote:
> Joe Conway escribió:
>
>> OK, how's this look?
>
> Hmm, is it possible to use OUT parameters in the function instead of
> declaring a new type for the result?

Sure, I guess I ought to use the latest-and-greatest. Any other comments
before I commit?

Thanks,

Joe
-----BEGIN PGP SIGNATURE-----
Version: GnuPG v1.4.9 (GNU/Linux)
Comment: Using GnuPG with Fedora - http://enigmail.mozdev.org

iQIcBAEBCAAGBQJKeDzfAAoJEDfy90M199hl3FsP/RYXWVVlItm3jk3hKNPCqTaT
maSwtl0QrFLl1pAc3ZTk16QdERrYFjfxIcxALz2am7OtStmhOz63Y4nlY8L0GHdz
ZJDF7K3r5o2RJ12h7Zucpo3racxp1eJqfi4S61RCeIi+aPxtcJd7dds1588FYACc
cPi6AuZyFfE6lhUjgniqOjzvTMptqWbtmuRZl1m+iyXsPE2FLlMclfugbzFKFXso
cnUI31cuejg9tF1zboatopm/qAcIiCB+U2KVG7tyiI+KdL/bNJlkELRJKODxdCvB
XNHzNBey1ZmbgE+H8gu1bz7DXbTxaffoQunCd4HnfOXOGrJpG+djsSCNW860sSVK
icCvdvq3taHfcpLtIIjpecu9LatTFGkJ12YVIV33gnv/Fgr8pj+84VsxRfqlfNaN
V743KdWtEv19gii3qIa817ZcS4tpBfIxXyt37cuZMdOv3VXR7LejeOoDr4aHGjDq
SCozcQQ+Fh8diU1XMzl3rRtz3Brz1cVYE8Ue8ELcQzzuOvWT+a0cjCBii1SmZdTz
DnpVEzmeOGrnHM8+IOG+h9IFRvNx3RQ2nxs71gcbgDp62fnQqb3m7c/Aj0xWv8xp
mQhY2DNQg0AY1vr71nfMGbYL7xTHFD5fA2Md2l/029PA+qFhJlqnZpWScHwqafcC
nkPH88kPUgy33vuNTJ3A
=U88M
-----END PGP SIGNATURE-----
Index: contrib/dblink/dblink.c
===================================================================
RCS file: /opt/src/cvs/pgsql/contrib/dblink/dblink.c,v
retrieving revision 1.82
diff -c -r1.82 dblink.c
*** contrib/dblink/dblink.c    11 Jun 2009 14:48:50 -0000    1.82
--- contrib/dblink/dblink.c    4 Aug 2009 13:41:26 -0000
***************
*** 1635,1640 ****
--- 1635,1723 ----
      PG_RETURN_DATUM(current_query(fcinfo));
  }

+ /*
+  * Retrieve async notifications for a connection.
+  *
+  * Returns an setof record of notifications, or an empty set if none recieved.
+  * Can optionally take a named connection as parameter, but uses the unnamed connection per default.
+  *
+  */
+ #define DBLINK_NOTIFY_COLS        3
+
+ PG_FUNCTION_INFO_V1(dblink_get_notify);
+ Datum
+ dblink_get_notify(PG_FUNCTION_ARGS)
+ {
+     PGconn               *conn = NULL;
+     remoteConn           *rconn = NULL;
+     PGnotify           *notify;
+     ReturnSetInfo       *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
+     TupleDesc            tupdesc;
+     Tuplestorestate       *tupstore;
+     MemoryContext        per_query_ctx;
+     MemoryContext        oldcontext;
+
+     DBLINK_INIT;
+     if (PG_NARGS() == 1)
+         DBLINK_GET_NAMED_CONN;
+     else
+         conn = pconn->conn;
+
+     /* create the tuplestore */
+     per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
+     oldcontext = MemoryContextSwitchTo(per_query_ctx);
+
+     tupdesc = CreateTemplateTupleDesc(DBLINK_NOTIFY_COLS, false);
+     TupleDescInitEntry(tupdesc, (AttrNumber) 1, "notify_name",
+                        TEXTOID, -1, 0);
+     TupleDescInitEntry(tupdesc, (AttrNumber) 2, "be_pid",
+                        INT4OID, -1, 0);
+     TupleDescInitEntry(tupdesc, (AttrNumber) 3, "extra",
+                        TEXTOID, -1, 0);
+
+     tupstore = tuplestore_begin_heap(true, false, work_mem);
+     rsinfo->returnMode = SFRM_Materialize;
+     rsinfo->setResult = tupstore;
+     rsinfo->setDesc = tupdesc;
+
+     MemoryContextSwitchTo(oldcontext);
+
+     PQconsumeInput(conn);
+     while ((notify = PQnotifies(conn)) != NULL)
+     {
+         Datum        values[DBLINK_NOTIFY_COLS];
+         bool        nulls[DBLINK_NOTIFY_COLS];
+
+         memset(values, 0, sizeof(values));
+         memset(nulls, 0, sizeof(nulls));
+
+         if (notify->relname != NULL)
+             values[0] = CStringGetTextDatum(notify->relname);
+         else
+             nulls[0] = true;
+
+         values[1] = Int32GetDatum(notify->be_pid);
+
+         if (notify->extra != NULL)
+             values[2] = CStringGetTextDatum(notify->extra);
+         else
+             nulls[2] = true;
+
+         /* switch to appropriate context while storing the tuple */
+         MemoryContextSwitchTo(per_query_ctx);
+         tuplestore_putvalues(tupstore, tupdesc, values, nulls);
+         MemoryContextSwitchTo(oldcontext);
+
+         PQfreemem(notify);
+         PQconsumeInput(conn);
+     }
+
+     /* clean up and return the tuplestore */
+     tuplestore_donestoring(tupstore);
+
+     return (Datum) 0;
+ }
+
  /*************************************************************
   * internal functions
   */
Index: contrib/dblink/dblink.h
===================================================================
RCS file: /opt/src/cvs/pgsql/contrib/dblink/dblink.h,v
retrieving revision 1.22
diff -c -r1.22 dblink.h
*** contrib/dblink/dblink.h    9 Jun 2009 17:41:02 -0000    1.22
--- contrib/dblink/dblink.h    4 Aug 2009 13:41:26 -0000
***************
*** 57,61 ****
--- 57,62 ----
  extern Datum dblink_build_sql_delete(PG_FUNCTION_ARGS);
  extern Datum dblink_build_sql_update(PG_FUNCTION_ARGS);
  extern Datum dblink_current_query(PG_FUNCTION_ARGS);
+ extern Datum dblink_get_notify(PG_FUNCTION_ARGS);

  #endif   /* DBLINK_H */
Index: contrib/dblink/dblink.sql.in
===================================================================
RCS file: /opt/src/cvs/pgsql/contrib/dblink/dblink.sql.in,v
retrieving revision 1.18
diff -c -r1.18 dblink.sql.in
*** contrib/dblink/dblink.sql.in    9 Jun 2009 17:41:02 -0000    1.18
--- contrib/dblink/dblink.sql.in    4 Aug 2009 13:44:34 -0000
***************
*** 202,204 ****
--- 202,223 ----
  RETURNS text
  AS 'MODULE_PATHNAME', 'dblink_error_message'
  LANGUAGE C STRICT;
+
+ CREATE OR REPLACE FUNCTION dblink_get_notify(
+     OUT notify_name TEXT,
+     OUT be_pid INT4,
+     OUT extra TEXT
+ )
+ RETURNS setof record
+ AS 'MODULE_PATHNAME', 'dblink_get_notify'
+ LANGUAGE C STRICT;
+
+ CREATE OR REPLACE FUNCTION dblink_get_notify(
+     conname TEXT,
+     OUT notify_name TEXT,
+     OUT be_pid INT4,
+     OUT extra TEXT
+ )
+ RETURNS setof record
+ AS 'MODULE_PATHNAME', 'dblink_get_notify'
+ LANGUAGE C STRICT;
Index: contrib/dblink/uninstall_dblink.sql
===================================================================
RCS file: /opt/src/cvs/pgsql/contrib/dblink/uninstall_dblink.sql,v
retrieving revision 1.7
diff -c -r1.7 uninstall_dblink.sql
*** contrib/dblink/uninstall_dblink.sql    5 Apr 2008 02:26:14 -0000    1.7
--- contrib/dblink/uninstall_dblink.sql    4 Aug 2009 13:41:26 -0000
***************
*** 76,78 ****
--- 76,82 ----
  DROP FUNCTION dblink_is_busy(text);

  DROP FUNCTION dblink_send_query(text, text);
+
+ DROP FUNCTION dblink_get_notify();
+
+ DROP FUNCTION dblink_get_notify(text);
Index: contrib/dblink/expected/dblink.out
===================================================================
RCS file: /opt/src/cvs/pgsql/contrib/dblink/expected/dblink.out,v
retrieving revision 1.25
diff -c -r1.25 dblink.out
*** contrib/dblink/expected/dblink.out    6 Jun 2009 21:27:56 -0000    1.25
--- contrib/dblink/expected/dblink.out    4 Aug 2009 13:41:26 -0000
***************
*** 827,829 ****
--- 827,870 ----
  DROP USER MAPPING FOR public SERVER fdtest;
  DROP SERVER fdtest;
  DROP FOREIGN DATA WRAPPER postgresql;
+ -- test asynchronous notifications
+ SELECT dblink_connect('dbname=contrib_regression');
+  dblink_connect
+ ----------------
+  OK
+ (1 row)
+
+ --should return listen
+ SELECT dblink_exec('LISTEN regression');
+  dblink_exec
+ -------------
+  LISTEN
+ (1 row)
+
+ --should return listen
+ SELECT dblink_exec('LISTEN foobar');
+  dblink_exec
+ -------------
+  LISTEN
+ (1 row)
+
+ NOTIFY regression;
+ NOTIFY foobar;
+ SELECT notify_name, be_pid/be_pid as be_pid, extra from dblink_get_notify();
+  notify_name | be_pid | extra
+ -------------+--------+-------
+  regression  |      1 |
+  foobar      |      1 |
+ (2 rows)
+
+ SELECT * from dblink_get_notify();
+  notify_name | be_pid | extra
+ -------------+--------+-------
+ (0 rows)
+
+ SELECT dblink_disconnect();
+  dblink_disconnect
+ -------------------
+  OK
+ (1 row)
+
Index: contrib/dblink/sql/dblink.sql
===================================================================
RCS file: /opt/src/cvs/pgsql/contrib/dblink/sql/dblink.sql,v
retrieving revision 1.21
diff -c -r1.21 dblink.sql
*** contrib/dblink/sql/dblink.sql    6 Jun 2009 21:27:56 -0000    1.21
--- contrib/dblink/sql/dblink.sql    4 Aug 2009 13:41:26 -0000
***************
*** 389,391 ****
--- 389,408 ----
  DROP USER MAPPING FOR public SERVER fdtest;
  DROP SERVER fdtest;
  DROP FOREIGN DATA WRAPPER postgresql;
+
+ -- test asynchronous notifications
+ SELECT dblink_connect('dbname=contrib_regression');
+
+ --should return listen
+ SELECT dblink_exec('LISTEN regression');
+ --should return listen
+ SELECT dblink_exec('LISTEN foobar');
+
+ NOTIFY regression;
+ NOTIFY foobar;
+
+ SELECT notify_name, be_pid/be_pid as be_pid, extra from dblink_get_notify();
+
+ SELECT * from dblink_get_notify();
+
+ SELECT dblink_disconnect();
Index: doc/src/sgml/dblink.sgml
===================================================================
RCS file: /opt/src/cvs/pgsql/doc/src/sgml/dblink.sgml,v
retrieving revision 1.8
diff -c -r1.8 dblink.sgml
*** doc/src/sgml/dblink.sgml    18 Jun 2009 14:34:36 -0000    1.8
--- doc/src/sgml/dblink.sgml    4 Aug 2009 13:41:26 -0000
***************
*** 1260,1265 ****
--- 1260,1338 ----
    </refsect1>
   </refentry>

+  <refentry id="CONTRIB-DBLINK-GET-NOTIFY">
+   <refnamediv>
+    <refname>dblink_get_notify</refname>
+    <refpurpose>retrieve async notifications on a connection</refpurpose>
+   </refnamediv>
+
+   <refsynopsisdiv>
+    <synopsis>
+     dblink_get_notify() returns setof (notify_name text, be_pid int, extra text)
+     dblink_get_notify(text connname) returns setof (notify_name text, be_pid int, extra text)
+    </synopsis>
+   </refsynopsisdiv>
+
+   <refsect1>
+    <title>Description</title>
+
+    <para>
+     <function>dblink_get_notify</> retrieves notifications on either
+     the unnamed connection, or on a named connection if specified.
+     To receive notifications via dblink, <function>LISTEN</> must
+     first be issued, using <function>dblink_exec</>.
+     For details see <xref linkend="sql-listen"> and <xref linkend="sql-notify">.
+    </para>
+
+   </refsect1>
+
+   <refsect1>
+    <title>Arguments</title>
+
+    <variablelist>
+     <varlistentry>
+      <term><parameter>conname</parameter></term>
+      <listitem>
+       <para>
+        The name of a named connection to get notifications on.
+       </para>
+      </listitem>
+     </varlistentry>
+    </variablelist>
+   </refsect1>
+
+   <refsect1>
+    <title>Return Value</title>
+     <para>Returns setof (notify_name text, be_pid int, extra text), or an empty set if none.</para>
+   </refsect1>
+
+   <refsect1>
+    <title>Example</title>
+
+    <programlisting>
+ test=# SELECT dblink_exec('LISTEN virtual');
+  dblink_exec
+ -------------
+  LISTEN
+ (1 row)
+
+ test=# SELECT * FROM dblink_get_notify();
+  notify_name | be_pid | extra
+ -------------+--------+-------
+ (0 rows)
+
+ test=# NOTIFY virtual;
+ NOTIFY
+
+ SELECT * FROM dblink_get_notify();
+  notify_name | be_pid | extra
+ -------------+--------+-------
+  virtual     |   1229 |
+ (1 row)
+    </programlisting>
+   </refsect1>
+  </refentry>
+
   <refentry id="CONTRIB-DBLINK-GET-RESULT">
    <refmeta>
     <refentrytitle>dblink_get_result</refentrytitle>