Refactor background psql TAP functions

This breaks out the background and interactive psql functionality into a
new class, PostgreSQL::Test::BackgroundPsql.  Sessions are still initiated
via PostgreSQL::Test::Cluster, but once started they can be manipulated by
the new helper functions which intend to make querying easier.  A sample
session for a command which can be expected to finish at a later time can
be seen below.

  my $session = $node->background_psql('postgres');
  $bsession->query_until(qr/start/, q(
    \echo start
	CREATE INDEX CONCURRENTLY idx ON t(a);
  ));
  $bsession->quit;

Patch by Andres Freund with some additional hacking by me.

Author: Andres Freund <andres@anarazel.de>
Reviewed-by: Andrew Dunstan <andrew@dunslane.net>
Discussion: https://postgr.es/m/20230130194350.zj5v467x4jgqt3d6@awork3.anarazel.de

1 files changed, 15 insertions, 36 deletions

diff --git a/src/test/subscription/t/015_stream.pl b/src/test/subscription/t/015_stream.pl
index 0e0f27f14df..88344bdbaa0 100644
--- a/src/test/subscription/t/015_stream.pl
+++ b/src/test/subscription/t/015_stream.pl
@@ -28,26 +28,20 @@ sub test_streaming
 	my ($node_publisher, $node_subscriber, $appname, $is_parallel) = @_;
 
 	# Interleave a pair of transactions, each exceeding the 64kB limit.
-	my $in  = '';
-	my $out = '';
-
 	my $offset = 0;
 
-	my $timer = IPC::Run::timeout($PostgreSQL::Test::Utils::timeout_default);
-
-	my $h = $node_publisher->background_psql('postgres', \$in, \$out, $timer,
+	my $h = $node_publisher->background_psql('postgres',
 		on_error_stop => 0);
 
 	# Check the subscriber log from now on.
 	$offset = -s $node_subscriber->logfile;
 
-	$in .= q{
+	$h->query_safe(q{
 	BEGIN;
 	INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
 	UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
 	DELETE FROM test_tab WHERE mod(a,3) = 0;
-	};
-	$h->pump_nb;
+	});
 
 	$node_publisher->safe_psql(
 		'postgres', q{
@@ -57,11 +51,9 @@ sub test_streaming
 	COMMIT;
 	});
 
-	$in .= q{
-	COMMIT;
-	\q
-	};
-	$h->finish;    # errors make the next test fail, so ignore them here
+	$h->query_safe('COMMIT');
+    # errors make the next test fail, so ignore them here
+	$h->quit;
 
 	$node_publisher->wait_for_catchup($appname);
 
@@ -219,12 +211,7 @@ $node_subscriber->reload;
 $node_subscriber->safe_psql('postgres', q{SELECT 1});
 
 # Interleave a pair of transactions, each exceeding the 64kB limit.
-my $in  = '';
-my $out = '';
-
-my $timer = IPC::Run::timeout($PostgreSQL::Test::Utils::timeout_default);
-
-my $h = $node_publisher->background_psql('postgres', \$in, \$out, $timer,
+my $h = $node_publisher->background_psql('postgres',
 	on_error_stop => 0);
 
 # Confirm if a deadlock between the leader apply worker and the parallel apply
@@ -232,11 +219,10 @@ my $h = $node_publisher->background_psql('postgres', \$in, \$out, $timer,
 
 my $offset = -s $node_subscriber->logfile;
 
-$in .= q{
+$h->query_safe(q{
 BEGIN;
 INSERT INTO test_tab_2 SELECT i FROM generate_series(1, 5000) s(i);
-};
-$h->pump_nb;
+});
 
 # Ensure that the parallel apply worker executes the insert command before the
 # leader worker.
@@ -246,11 +232,8 @@ $node_subscriber->wait_for_log(
 
 $node_publisher->safe_psql('postgres', "INSERT INTO test_tab_2 values(1)");
 
-$in .= q{
-COMMIT;
-\q
-};
-$h->finish;
+$h->query_safe('COMMIT');
+$h->quit;
 
 $node_subscriber->wait_for_log(qr/ERROR: ( [A-Z0-9]+:)? deadlock detected/,
 	$offset);
@@ -277,11 +260,10 @@ $node_subscriber->safe_psql('postgres',
 # Check the subscriber log from now on.
 $offset = -s $node_subscriber->logfile;
 
-$in .= q{
+$h->query_safe(q{
 BEGIN;
 INSERT INTO test_tab_2 SELECT i FROM generate_series(1, 5000) s(i);
-};
-$h->pump_nb;
+});
 
 # Ensure that the first parallel apply worker executes the insert command
 # before the second one.
@@ -292,11 +274,8 @@ $node_subscriber->wait_for_log(
 $node_publisher->safe_psql('postgres',
 	"INSERT INTO test_tab_2 SELECT i FROM generate_series(1, 5000) s(i)");
 
-$in .= q{
-COMMIT;
-\q
-};
-$h->finish;
+$h->query_safe('COMMIT');
+$h->quit;
 
 $node_subscriber->wait_for_log(qr/ERROR: ( [A-Z0-9]+:)? deadlock detected/,
 	$offset);