/*
* Copyright (c) 2015-2024 Tada AB and other contributors, as listed below.
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the The BSD 3-Clause License
* which accompanies this distribution, and is available at
* http://opensource.org/licenses/BSD-3-Clause
*
* Contributors:
* Chapman Flack (this file, 2020)
* PostgreSQL Global Development Group, Michael Paquier, Alvaro Herrera
* (PostgresNode.pm, 2015, of which similar methods here are ports)
*/
package org.postgresql.pljava.packaging;
import org.gjt.cuspy.JarX;
import java.io.InputStream;
import static java.lang.System.getProperty;
import static java.lang.System.setProperty;
import java.nio.ByteBuffer;
import static java.nio.charset.Charset.defaultCharset;
import java.util.regex.Matcher;
import static java.util.regex.Pattern.compile;
/*
* For "Node" behavior:
*/
import static java.lang.ProcessBuilder.Redirect.INHERIT;
import java.lang.reflect.InvocationHandler; // flexible SAM allowing exceptions
import java.lang.reflect.UndeclaredThrowableException;
import static java.lang.Thread.interrupted;
import java.lang.invoke.MethodHandle;
import java.lang.invoke.MethodHandles.Lookup;
import static java.lang.invoke.MethodHandles.explicitCastArguments;
import static java.lang.invoke.MethodHandles.filterReturnValue;
import static java.lang.invoke.MethodHandles.publicLookup;
import static java.lang.invoke.MethodType.methodType;
import static java.net.InetAddress.getLoopbackAddress;
import static java.net.URLEncoder.encode;
import java.net.ServerSocket;
import static java.nio.charset.StandardCharsets.US_ASCII;
import static java.nio.file.Files.createTempFile;
import static java.nio.file.Files.createTempDirectory;
import static java.nio.file.Files.deleteIfExists;
import static java.nio.file.Files.exists;
import static java.nio.file.Files.getLastModifiedTime;
import static java.nio.file.Files.lines;
import static java.nio.file.Files.walk;
import static java.nio.file.Files.write;
import java.nio.file.Path;
import java.nio.file.Paths;
import static java.nio.file.StandardWatchEventKinds.*;
import java.nio.file.WatchEvent;
import java.nio.file.WatchKey;
import java.nio.file.WatchService;
import java.nio.file.AccessDeniedException;
import java.nio.file.NoSuchFileException;
import java.sql.Connection;
import static java.sql.DriverManager.drivers;
import static java.sql.DriverManager.getConnection;
import java.sql.ParameterMetaData;
import java.sql.PreparedStatement;
import java.sql.Statement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.Types;
import java.sql.SQLException;
import java.sql.SQLWarning;
import javax.sql.rowset.RowSetProvider;
import javax.sql.rowset.WebRowSet;
import javax.sql.rowset.RowSetMetaDataImpl;
import java.util.ArrayDeque;
import java.util.Base64;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
import java.util.Objects;
import static java.util.Objects.requireNonNull;
import java.util.Properties;
import java.util.Random;
import java.util.Spliterator;
import static java.util.Spliterator.IMMUTABLE;
import static java.util.Spliterator.NONNULL;
import static java.util.Spliterator.ORDERED;
import static java.util.Spliterators.spliteratorUnknownSize;
import java.util.WeakHashMap;
import java.util.concurrent.Callable; // like a Supplier but allows exceptions!
import java.util.concurrent.CancellationException;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import java.util.function.BooleanSupplier;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.Supplier;
import java.util.function.UnaryOperator;
import java.util.jar.JarFile;
import java.util.stream.IntStream;
import java.util.stream.Stream;
import static java.util.stream.StreamSupport.stream;
/**
* Extends the JarX extraction tool to provide a {@code resolve} method that
* replaces prefixes {@code pljava/foo/} in path names stored in the archive
* with the result of {@code pg_config --foo}.
*
* As this represents a second extra {@code .class} file that has to be added
* to the installer jar anyway, it will also contain some methods intended to be
* useful for tasks related to installation and testing. The idea is not to go
* overboard, but supply a few methods largely modeled on the most basic ones of
* PostgreSQL's {@code PostgreSQL::Test::Cluster} Perl module (formerly named
* {@code PostgresNode}, from which the name of this class was taken). The
* methods can be invoked from {@code jshell} if its classpath includes the
* installer jar (and one of the PostgreSQL JDBC drivers).
*
* Unlike the many capabilities of {@code PostgreSQL::Test::Cluster}, this only
* deals in TCP sockets bound to {@code localhost}
* ({@code StandardProtocolFamily.UNIX}
* finally arrived in Java 16 but this class does not support it yet) and only
* a few of the most basic operations.
*
* As in JarX itself, some liberties with coding style may be taken here to keep
* this one extra {@code .class} file from proliferating into a bunch of them.
*
* As the testing-related methods here are intended for ad-hoc or scripted use
* in {@code jshell}, they are typically declared to throw any checked
* exception, without further specifics. There are many overloads of methods
* named {@code q} and {@code qp} (mnemonic of query and query-print), to make
* interactive use in {@code jshell} comfortable with just a few static imports.
*/
public class Node extends JarX {
private Matcher m_prefix;
private int m_fsepLength;
private String m_lineSep;
private boolean m_dryrun = false;
private static Node s_jarxHelper = new Node(null, 0, null, null);
private static boolean s_jarProcessed = false;
private static String s_examplesJar;
private static String s_sharedObject;
/**
* Performs an ordinary installation, using {@code pg_config} or the
* corresponding system properties to learn where the files belong, and
* unpacking the files (not including this class or its ancestors) there.
*/
public static void main(String[] args) throws Exception
{
if ( args.length > 0 )
{
System.err.println("usage: java -jar filename.jar");
System.exit(1);
}
s_jarxHelper.extract();
}
/**
* Extracts the jar contents, just as done in the normal case of running
* this class with {@code java -jar}.
*
* Only to be called on the singleton instance {@code s_jarxHelper}.
*
* For a version that doesn't really extract anything, but still primes the
* {@code resolve} method to know where things should be extracted,
* see {@link #dryExtract}.
*/
@Override
public void extract() throws Exception
{
super.extract();
s_jarProcessed = true;
}
/**
* Prepares the resolver, ignoring the passed string (ordinarily a script or
* rules); this resolver's rules are hardcoded.
*/
@Override
public void prepareResolver(String v) throws Exception
{
m_prefix = compile("^pljava/([^/]+dir)(?![^/])").matcher("");
m_fsepLength = getProperty("file.separator").length();
m_lineSep = getProperty("line.separator");
}
/**
* Replaces a prefix {@code pljava/}key in a path to be extracted
* with the value of the {@code pgconfig.}key system property, or
* the result of invoking {@code pg_config} (or the exact executable named
* in the {@code pgconfig} system property, if present) with the option
* {@code --}key.
*/
@Override
public String resolve(String storedPath, String platformPath)
throws Exception
{
if ( m_prefix.reset(storedPath).lookingAt() )
{
int prefixLength = m_prefix.end();
String key = m_prefix.group(1);
String propkey = "pgconfig." + key;
String replacement = getProperty(propkey);
if ( null == replacement )
{
String pgc = getProperty("pgconfig", "pg_config");
ProcessBuilder pb = new ProcessBuilder(pgc, "--"+key);
pb.redirectError(ProcessBuilder.Redirect.INHERIT);
Process proc = pb.start();
byte[] output;
try ( InputStream instream = proc.getInputStream() )
{
proc.getOutputStream().close();
output = instream.readAllBytes();
}
finally
{
int status = proc.waitFor();
if ( 0 != status )
{
System.err.println(
"ERROR: pg_config status is "+status);
System.exit(1);
}
}
/*
* pg_config output is the saved value followed by one \n only.
* However, on Windows, the C library treats stdout as text mode
* by default, and pg_config does nothing to change that, so the
* single \n written by pg_config gets turned to \r\n before it
* arrives here. The earlier use of the trim() method papered
* over the problem, but trim() can remove too much. Simply have
* to assume that the string will end with line.separator, and
* remove that.
*/
replacement = defaultCharset().newDecoder()
.decode(ByteBuffer.wrap(output, 0, output.length))
.toString();
assert replacement.endsWith(m_lineSep);
replacement = replacement.substring(0,
replacement.length() - m_lineSep.length());
setProperty(propkey, replacement);
}
int plen = m_fsepLength - 1; /* original separator had length 1 */
plen += prefixLength;
replacement += platformPath.substring(plen);
if ( -1 != storedPath.indexOf("/pljava-examples-") )
s_examplesJar = replacement;
else if ( storedPath.matches(
"pljava/pkglibdir/(?:lib)?+pljava-so-.*") )
s_sharedObject = replacement;
if ( ! m_dryrun )
return replacement;
return null;
}
System.err.println("WARNING: extraneous jar entry not extracted: "
+ storedPath);
return null;
}
/*
* Members below this point represent the state and behavior of an instance
* of this class that is acting as a "Node" rather than as the JarX helper.
*/
/**
* True if the platform is determined to be Windows.
*
* On Windows, {@link #forWindowsCRuntime forWindowsCRuntime} should be
* applied to any {@code ProcessBuilder} before invoking it; the details of
* the transformation applied by
* {@link #asPgCtlInvocation asPgCtlInvocation} change, and
* {@link #use_pg_ctl use_pg_ctl} may prove useful, as {@code pg_ctl} on
* Windows is able to drop administrative privileges that would otherwise
* prevent {@code postgres} from starting.
*/
public static final boolean s_isWindows =
getProperty("os.name").startsWith("Windows");
/**
* The first form of PostgreSQL JDBC driver connection URL found to be
* recognized by an available driver, or {@code URL_FORM_NONE}.
*/
public static final int s_urlForm;
/**
* Value of {@link #s_urlForm s_urlForm} indicating no available JDBC driver
* was found to accept any of the supported connection URL forms.
*/
public static final int URL_FORM_NONE = -1;
/**
* Value of {@link #s_urlForm s_urlForm} indicating an available JDBC driver
* reported accepting a connection URL in the PGJDBC form starting with
* {@code "jdbc:postgresql:"}.
*/
public static final int URL_FORM_PGJDBC = 0;
/**
* Value of {@link #s_urlForm s_urlForm} indicating an available JDBC driver
* reported accepting a connection URL in the pgjdbc-ng form starting with
* {@code "jdbc:pgsql:"}.
*/
public static final int URL_FORM_PGJDBCNG = 1;
/**
* A function to map an {@code SQLWarning} to a rough classification
* (info, warning) of its severity.
*
* If the PGJDBC {@code PSQLWarning} class is available for access to the
* severity tag from the backend, "warning" will be returned if that tag is
* {@code WARNING}, and "info" will be returned in any other case. (The next
* more severe backup level is {@code ERROR}, which would not appear here as
* an {@code SQLWarning}.)
*
* If the severity tag is not available, "info" will be returned if the
* class (leftmost two positions of SQLState) is 00, otherwise "warning".
*/
private static final Function s_toSeverity;
private static String s_WARNING_localized = "WARNING";
/**
* Changes the severity string used to recognize when the backend is sending
* a {@code WARNING}.
*
* When the driver is PGJDBC, the classification done here of
* {@code SQLWarning} instances into actual warning messages or informative
* ones depends on a tag ("WARNING" in English) that the backend delivers
* in the local language. For the classification to happen correctly when
* a different language is selected, use this method to supply the string
* (for example, "PERINGATAN" in Indonesian) that the backend uses for
* warnings in that language.
*/
public static void set_WARNING_localized(String s)
{
s_WARNING_localized = requireNonNull(s);
}
static
{
String[] candidateURLs = { "jdbc:postgresql:", "jdbc:pgsql:x" };
s_urlForm =
IntStream.range(0, candidateURLs.length)
.filter(i ->
drivers().anyMatch(d ->
{
try
{
return d.acceptsURL(candidateURLs[i]);
}
catch ( SQLException e )
{
throw new ExceptionInInitializerError(e);
}
}))
.findFirst()
.orElse(URL_FORM_NONE);
Function toSeverity = Node::toSeverityFallback;
try
{
Class> psqlWarning =
Class.forName("org.postgresql.util.PSQLWarning");
Class> sErrMessage =
Class.forName("org.postgresql.util.ServerErrorMessage");
Lookup pub = publicLookup();
MethodHandle getserrm =
pub.findVirtual(psqlWarning, "getServerErrorMessage",
methodType(sErrMessage));
MethodHandle getSev =
pub.findVirtual(sErrMessage, "getSeverity",
methodType(String.class));
MethodHandle h = explicitCastArguments(
filterReturnValue(getserrm, getSev),
methodType(String.class, Object.class));
toSeverity = w ->
{
if ( psqlWarning.isInstance(w) )
{
try
{
String s = (String)h.invokeExact(psqlWarning.cast(w));
if ( null == s || s_WARNING_localized.equals(s) )
return "warning";
return "info";
}
catch ( Throwable t )
{
throw new UndeclaredThrowableException(t, t.getMessage());
}
}
return toSeverityFallback(w);
};
}
catch ( ReflectiveOperationException e )
{
}
s_toSeverity = toSeverity;
}
private static String toSeverityFallback(SQLWarning w)
{
if ( w.getSQLState().startsWith("00") )
return "info";
else
return "warning";
}
/**
* A state (see {@link #stateMachine stateMachine}) that expects nothing
* (if the driver is pgjdbc-ng) or a zero row count (if the driver is
* PGJDBC).
*
* For some utility statements (such as {@code CREATE EXTENSION}) with no
* result, the pgjdbc-ng driver will produce no result, while the PGJDBC
* driver produces a zero count, as it would for a DML statement that did
* not affect any rows. This state handles either case.
*
* When {@code URL_FORM_PGJDBCNG == s_urlForm}, this state consumes nothing
* and moves to the numerically next state. Otherwise (JDBC), it checks
* that the current object is a zero row count, consuming it and moving to
* the numerically next state if it is, returning false otherwise.
*/
public static final InvocationHandler NOTHING_OR_PGJDBC_ZERO_COUNT=(o,p,q)->
{
int myStateNum = (int)q[0];
if ( URL_FORM_PGJDBCNG == s_urlForm )
return -(1 + myStateNum);
return 0 == as(Long.class, o) ? 1 + myStateNum : false;
};
/**
* Name of a "Node"; null for an ordinary Node instance.
*/
private final String m_name;
/**
* A TCP port on {@code localhost} that was free when {@code get_new_node}
* was called, and is likeliest to still be free if {@code start} is then
* called without undue delay.
*/
private final int m_port;
/**
* A temporary base directory chosen and created in {@code java.io.tmpdir}
* by {@code get_new_node} and removed by {@code clean_node}.
*/
private final Path m_basedir;
/**
* A password generated at {@code get_new_node} time, and used by
* {@code init} as the database-superuser password passed to {@code initdb}.
*/
private final String m_password;
/**
* The server process handle after a successful {@code start}
* via {@code pg_ctl}; null again after a successful {@code stop}.
*
* If {@code pg_ctl} was not used, this will be null and {@code m_server}
* will have a value.
*/
private ProcessHandle m_serverHandle;
/**
* The server process after a successful {@code start}; null again after a
* successful {@code stop}.
*
* If {@code pg_ctl} was used to start the server, this will be null and
* {@code m_serverHandle} will have a value after {@code wait_for_pid_file}.
*/
private Process m_server;
/**
* A count of connections, used to supply a distinct default
* {@code ApplicationName} per connection.
*/
private long m_connCount = 0;
/**
* Whether to invoke {@code postgres} directly when starting the server,
* or use {@code pg_ctl} to start and stop it.
*
* On Windows, {@code pg_ctl} is able to drop administrator rights and
* start the server from an account that would otherwise trigger
* the server's refusal to start from a privileged account.
*/
private boolean m_usePostgres = true;
/**
* A weakly-held collection of {@link Connection}s, so that any remaining
* unclosed when {@link #stop(UnaryOperator) stop} is called can be closed
* then.
*
* Java takes care of removing {@code Connection}s from this map as they
* become unreachable. In case any become unreachable before being closed,
* both supported JDBC drivers have cleaner actions that will eventually
* close them.
*/
private final WeakHashMap m_connections;
/**
* True during a {@link #stop(UnaryOperator) stop} call.
*
* Used to prevent any new unclosed {@code Connection} being added to
* {@link m_connections m_connections} undetected.
*/
private boolean m_stopping = false;
/**
* Identifying information for a "node" instance, or for the singleton
* extractor instance.
*/
@Override
public String toString()
{
if ( null == m_name )
return "Extractor instance";
return "\"Node\": " + m_name;
}
/**
* Constructs an instance; all nulls for the parameters are passed by the
* static initializer to make the singleton extractor instance, and any
* other instance is constructed by {@code get_new_node} for controlling
* a PostgreSQL instance.
*/
private Node(String nodeName, int port, Path basedir, String password)
{
m_name = nodeName;
m_port = port;
m_basedir = basedir;
m_password = password;
m_connections = null == nodeName ? null : new WeakHashMap<>();
}
/**
* Returns a new {@code Node} that can be used to initialize and start a
* PostgreSQL instance.
*
* Establishes a VM shutdown hook that will stop the server (if started)
* and recursively remove the basedir before the VM exits.
*/
public static Node get_new_node(String name) throws Exception
{
byte[] pwbytes = new byte [ 6 ];
new Random().nextBytes(pwbytes);
Node n = new Node(
requireNonNull(name),
get_free_port(),
createTempDirectory("t_pljava_" + name + "_data"),
Base64.getEncoder().encodeToString(pwbytes));
Thread t =
new Thread(() ->
{
try
{
n.stop();
n.clean_node();
}
catch ( Exception e )
{
e.printStackTrace();
}
}, "Node " + name + " shutdown");
Runtime.getRuntime().addShutdownHook(t);
return n;
}
/**
* Returns a TCP port on the loopback interface that is free at the moment
* this method is called.
*/
public static int get_free_port() throws Exception
{
try (ServerSocket s = new ServerSocket(0, 0, getLoopbackAddress()))
{
return s.getLocalPort();
}
}
/**
* Recursively removes the basedir and its descendants.
*/
public void clean_node() throws Exception
{
clean_node(false);
}
/**
* Recursively removes the basedir (unless keepRoot)
* and its descendants.
* @param keepRoot if true, the descendants are removed, but not the basedir
* itself.
*/
public void clean_node(boolean keepRoot) throws Exception
{
/*
* How can Java *still* not have a deleteTree()?
*/
ArrayDeque stk = new ArrayDeque<>();
for ( Path p : (Iterable)walk(m_basedir)::iterator )
{
while ( ! stk.isEmpty() && ! p.startsWith(stk.peek()) )
{
Path toDelete = stk.pop();
try
{
deleteIfExists(toDelete);
}
catch ( AccessDeniedException e )
{
if (!toDelete.equals(data_dir().resolve("postmaster.pid")))
throw e;
/*
* See comments for stopViaPgCtl regarding this weirdness.
*/
Thread.sleep(500);
deleteIfExists(toDelete);
}
}
stk.push(p);
}
if ( keepRoot )
stk.pollLast();
for ( Path p : stk )
deleteIfExists(p);
}
/**
* Processes the jar without really extracting, to compute
* the path mappings.
*/
private static void dryExtract() throws Exception
{
if ( s_jarProcessed )
return;
try
{
s_jarxHelper.m_dryrun = true;
s_jarxHelper.extract();
}
finally
{
s_jarxHelper.m_dryrun = false;
}
}
/**
* Given a path from the archive, or any path resembling one in
* the archive (that is, always {@code /} as the separator, and starting
* with {@code pljava/}key where {@code --}key is known
* to {@code pg_config}, returns the platform-specific path where it would
* be installed.
*/
private static String resolve(String archivePath) throws Exception
{
return s_jarxHelper.resolve(
archivePath, Paths.get("", archivePath.split("/")).toString());
}
/**
* Returns the directory name to be used as the PostgreSQL data directory
* for this node.
*/
public Path data_dir()
{
return m_basedir.resolve("pgdata");
}
/**
* Like {@code init()} but returns an {@code AutoCloseable} that will
* recursively remove the files and directories under the basedir
* (but not the basedir itself) on the exit of a calling
* try-with-resources scope.
*/
public AutoCloseable initialized_cluster()
throws Exception
{
return initialized_cluster(Map.of(), UnaryOperator.identity());
}
/**
* Like {@code init()} but returns an {@code AutoCloseable} that will
* recursively remove the files and directories under the basedir
* (but not the basedir itself) on the exit of a calling
* try-with-resources scope.
*/
public AutoCloseable initialized_cluster(Map suppliedOptions)
throws Exception
{
return initialized_cluster(suppliedOptions, UnaryOperator.identity());
}
/**
* Like {@link #init(Map,UnaryOperator) init()} but returns
* an {@code AutoCloseable} that will
* recursively remove the files and directories under the basedir
* (but not the basedir itself) on the exit of a calling
* try-with-resources scope.
*/
public AutoCloseable initialized_cluster(
UnaryOperator tweaks)
throws Exception
{
return initialized_cluster(Map.of(), tweaks);
}
/**
* Like {@link #init(Map,UnaryOperator) init()} but returns
* an {@code AutoCloseable} that will
* recursively remove the files and directories under the basedir
* (but not the basedir itself) on the exit of a calling
* try-with-resources scope.
*/
public AutoCloseable initialized_cluster(
Map suppliedOptions,
UnaryOperator tweaks)
throws Exception
{
init(suppliedOptions, tweaks);
return () ->
{
clean_node(true);
};
}
/**
* Invokes {@code initdb} for the node, passing default options appropriate
* for this setting.
*/
public void init() throws Exception
{
init(Map.of(), UnaryOperator.identity());
}
/**
* Invokes {@code initdb} for the node, with suppliedOptions
* overriding or supplementing the ones that would be passed by default.
*/
public void init(Map suppliedOptions) throws Exception
{
init(suppliedOptions, UnaryOperator.identity());
}
/**
* Invokes {@code initdb} for the node, passing default options appropriate
* for this setting, and {@linkplain #init(Map,UnaryOperator) tweaks} to be
* applied to the {@code ProcessBuilder} before it is started.
*/
public void init(UnaryOperator tweaks) throws Exception
{
init(Map.of(), tweaks);
}
/**
* Invokes {@code initdb} for the node, with suppliedOptions
* overriding or supplementing the ones that would be passed by default,
* and tweaks to be applied to the {@code ProcessBuilder}
* before it is started.
*
* By default, {@code postgres} will be the name of the superuser, UTF-8
* will be the encoding, {@code auth-local} will be {@code peer} and
* {@code auth-host} will be {@code md5}. The initialization will skip
* {@code fsync} for speed rather than safety (if something goes wrong, just
* {@code clean_node()} and start over).
*
* The {@code initdb} that will be run is the one in the {@code bindir}
* reported by {@code pg_config} (or set by {@code -Dpgconfig.bindir}).
* @param suppliedOptions a Map where each key is an option to initdb
* (for example, --encoding), and the value corresponds.
* @param tweaks a lambda applicable to the {@code ProcessBuilder} to
* further configure it. On Windows, the tweaks will be applied ahead of
* transformation of the arguments by
* {@link #forWindowsCRuntime forWindowsCRuntime}.
*/
public void init(
Map suppliedOptions,
UnaryOperator tweaks) throws Exception
{
dryExtract();
/*
* For extract/install purposes, there is already a resolve() method
* that expands keys like pljava/bindir to pg_config --bindir output.
*/
String initdb = resolve("pljava/bindir/initdb");
if ( s_isWindows )
{
/*
* This is irksome. The mingw64 postgresql package has both
* initdb.exe and initdb, a bash script that runs it under winpty.
* If the script were not there, the .exe suffix would be added
* implicitly, but with both there, we try to exec the bash script.
*/
Path p1 = Paths.get(initdb);
Path p2 = Paths.get(initdb + ".exe");
if ( exists(p1) && exists(p2) )
initdb = p2.toString();
}
Path pwfile = createTempFile(m_basedir, "pw", "");
Map options = new HashMap<>(suppliedOptions);
options.putIfAbsent("--pgdata", data_dir().toString());
options.putIfAbsent("--username", "postgres");
options.putIfAbsent("--encoding", "utf-8");
options.putIfAbsent("--pwfile", pwfile.toString());
options.putIfAbsent("--auth-local", "peer");
options.putIfAbsent("--auth-host", "md5");
options.putIfAbsent("-N", null);
String[] args =
Stream.concat(
Stream.of(initdb),
options.entrySet().stream()
.flatMap(e ->
null == e.getValue()
? Stream.of(e.getKey())
: Stream.of(e.getKey(), e.getValue()))
)
.toArray(String[]::new);
try
{
write(pwfile, List.of(m_password), US_ASCII);
ProcessBuilder pb =
new ProcessBuilder(args)
.redirectOutput(INHERIT)
.redirectError(INHERIT);
pb = tweaks.apply(pb);
if ( s_isWindows )
pb = forWindowsCRuntime(pb);
Process p = pb.start();
p.getOutputStream().close();
if ( 0 != p.waitFor() )
throw new AssertionError(
"Nonzero initdb result: " + p.waitFor());
}
finally
{
deleteIfExists(pwfile);
}
}
/**
* Like {@code start()} but returns an {@code AutoCloseable} that will
* stop the server on the exit of a calling try-with-resources scope.
*/
public AutoCloseable started_server()
throws Exception
{
return started_server(Map.of(), UnaryOperator.identity());
}
/**
* Like {@code start()} but returns an {@code AutoCloseable} that will
* stop the server on the exit of a calling try-with-resources scope.
*/
public AutoCloseable started_server(Map suppliedOptions)
throws Exception
{
return started_server(suppliedOptions, UnaryOperator.identity());
}
/**
* Like {@link #start(Map,UnaryOperator) start()} but returns
* an {@code AutoCloseable} that will
* stop the server on the exit of a calling try-with-resources scope.
*
* Supplied tweaks will be applied to the {@code ProcessBuilder}
* used to start the server; if {@code pg_ctl} is being used, they will also
* be applied when running {@code pg_ctl stop} to stop it.
*/
public AutoCloseable started_server(UnaryOperator tweaks)
throws Exception
{
return started_server(Map.of(), tweaks);
}
/**
* Like {@link #start(Map,UnaryOperator) start()} but returns
* an {@code AutoCloseable} that will
* stop the server on the exit of a calling try-with-resources scope.
*
* Supplied tweaks will be applied to the {@code ProcessBuilder}
* used to start the server; if {@code pg_ctl} is being used, they will also
* be applied when running {@code pg_ctl stop} to stop it.
*/
public AutoCloseable started_server(
Map suppliedOptions,
UnaryOperator tweaks)
throws Exception
{
start(suppliedOptions, tweaks);
return () ->
{
stop(tweaks);
};
}
/**
* Starts a PostgreSQL server for the node with default options appropriate
* for this setting.
*/
public void start() throws Exception
{
start(Map.of(), UnaryOperator.identity());
}
/**
* Starts a PostgreSQL server for the node, with suppliedOptions
* overriding or supplementing the ones that would be passed by default.
*/
public void start(Map suppliedOptions) throws Exception
{
start(suppliedOptions, UnaryOperator.identity());
}
/**
* Starts a PostgreSQL server for the node, passing default options
* appropriate for this setting, and
* {@linkplain #start(Map,UnaryOperator) tweaks} to be
* applied to the {@code ProcessBuilder} before it is started.
*/
public void start(UnaryOperator tweaks) throws Exception
{
start(Map.of(), tweaks);
}
/**
* Starts a PostgreSQL server for the node, with suppliedOptions
* overriding or supplementing the ones that would be passed by default, and
* tweaks to be applied to the {@code ProcessBuilder} before it
* is started.
*
* By default, the server will listen only on the loopback interface and
* not on any Unix-domain socket, on the port selected when this Node was
* created, and for a maximum of 16 connections. Its cluster name will be
* the name given to this Node, and fsync will be off to favor speed over
* durability. The log line prefix will be shortened to just the node name
* and (when connected) the {@code application_name}.
*
* The server that will be run is the one in the {@code bindir}
* reported by {@code pg_config} (or set by {@code -Dpgconfig.bindir}).
*
* If the server is PostgreSQL 10 or later, it is definitely ready to accept
* connections when this method returns. If not, it is highly likely to be
* ready, but no test connection has been made to confirm it.
* @param suppliedOptions a Map where the key is a configuration variable
* name as seen in {@code postgresql.conf} or passed to the server with
* {@code -c} and the value corresponds.
* @param tweaks a lambda applicable to the {@code ProcessBuilder} to
* further configure it. Under {@link #use_pg_ctl use_pg_ctl(true)}, the
* tweaks are applied after the arguments have been transformed by
* {@link #asPgCtlInvocation asPgCtlInvocation}. On Windows, they are
* applied ahead of transformation of the arguments by
* {@link #forWindowsCRuntime forWindowsCRuntime}.
*/
public void start(
Map suppliedOptions,
UnaryOperator tweaks) throws Exception
{
if ( null != m_server && m_server.isAlive() )
throw new IllegalStateException(
"node \"" + m_name + "\" is already running");
if ( null != m_serverHandle && m_serverHandle.isAlive() )
throw new IllegalStateException(
"node \"" + m_name + "\" is already running");
dryExtract();
Stream cmd = Stream.of(resolve("pljava/bindir/postgres"));
Map options = new HashMap<>(suppliedOptions);
options.putIfAbsent("data_directory", data_dir().toString());
options.putIfAbsent("listen_addresses",
getLoopbackAddress().getHostAddress());
options.putIfAbsent("port", "" + m_port);
options.putIfAbsent("unix_socket_directories", "");
options.putIfAbsent("max_connections", "16");
options.putIfAbsent("fsync", "off");
options.putIfAbsent("cluster_name", m_name);
options.putIfAbsent("log_line_prefix",
m_name.replace("%", "%%") + ":%q%a:");
String[] args =
Stream.concat(
cmd,
options.entrySet().stream()
.flatMap(e ->
"data_directory".equals(e.getKey())
? Stream.of("-D", e.getValue())
: Stream.of("-c", e.getKey() + "=" + e.getValue())
)
)
.toArray(String[]::new);
ProcessBuilder pb =
new ProcessBuilder(args)
.redirectOutput(INHERIT)
.redirectError(INHERIT);
if ( ! m_usePostgres )
pb = asPgCtlInvocation(pb);
pb = tweaks.apply(pb);
if ( s_isWindows )
pb = forWindowsCRuntime(pb);
Process p = pb.start();
p.getOutputStream().close();
try
{
wait_for_pid_file(p, p.info());
if ( m_usePostgres )
m_server = p; // else wait_for_pid_file has set m_serverHandle
}
finally
{
if ( m_server == p )
return;
if ( p.isAlive() )
p.destroy();
}
}
/**
* Stops the server instance associated with this Node.
*
* Has the effect of {@link #stop(UnaryOperator) stop(tweaks)} without
* any tweaks.
*/
public void stop() throws Exception
{
stop(UnaryOperator.identity());
}
/**
* Stops the server instance associated with this Node.
*
* No effect if it has not been started or has already been stopped, but
* a message to standard error is logged if the server had been started and
* the process is found to have exited unexpectedly.
* @param tweaks tweaks to apply to a ProcessBuilder; unused unless
* {@code pg_ctl} will be used to stop the server. When used, they are
* applied ahead of the transformation of the arguments by
* {@link #forWindowsCRuntime forWindowsCRuntime} used on Windows.
*/
public void stop(UnaryOperator tweaks) throws Exception
{
if ( null == ( m_usePostgres ? m_server : m_serverHandle ) )
return;
try
{
Connection[] connections;
synchronized ( this )
{
m_stopping = true;
connections = // Java >= 10: use a List and List.copyOf
m_connections.keySet().stream().toArray(Connection[]::new);
m_connections.clear();
}
for ( Connection c : connections )
{
try
{
c.close();
}
catch ( Exception e )
{
}
}
if ( ! m_usePostgres )
{
stopViaPgCtl(tweaks);
return;
}
if ( m_server.isAlive() )
{
m_server.destroy();
m_server.waitFor();
m_server = null;
return;
}
System.err.println("Server had already exited with status " +
m_server.exitValue());
m_server = null;
}
finally
{
synchronized ( this )
{
m_stopping = false;
}
}
}
private void stopViaPgCtl(UnaryOperator tweaks)
throws Exception
{
if ( ! m_serverHandle.isAlive() )
{
System.err.println("Server had already exited");
m_serverHandle = null;
return;
}
String pg_ctl = resolve("pljava/bindir/pg_ctl");
ProcessBuilder pb = new ProcessBuilder(
pg_ctl, "stop", "-D", data_dir().toString(), "-m", "fast")
.redirectOutput(INHERIT)
.redirectError(INHERIT);
pb = tweaks.apply(pb);
if ( s_isWindows )
pb = forWindowsCRuntime(pb);
Process p = pb.start();
p.getOutputStream().close();
if ( 0 != p.waitFor() )
{
/*
* Here is a complication. On Windows, pg_ctl suffers from a race
* condition that can occasionally cause it to exit with a nonzero
* status and a "permission denied" message about postmaster.pid,
* while the server is otherwise successfully stopped:
* www.postgresql.org/message-id/16922.1520722108%40sss.pgh.pa.us
*
* Without capturing the stderr of the process (too much bother), we
* won't know for sure if that is the message, but if the exit value
* was nonzero, just wait a bit and see if the server has gone away;
* if it has, don't worry about it.
*/
Thread.sleep(1000);
if ( m_serverHandle.isAlive() )
throw new AssertionError(
"Nonzero pg_ctl stop result: " + p.waitFor());
}
m_serverHandle = null;
}
/**
* Sets whether to use {@code pg_ctl} to start and stop the server
* (if true), or start {@code postgres} and stop it directly (if false,
* the default).
*
* On Windows, {@code pg_ctl} is able to drop administrator rights and
* start the server from an account that would otherwise trigger
* the server's refusal to start from a privileged account.
*/
public void use_pg_ctl(boolean setting)
{
if ( null != m_server || null != m_serverHandle )
throw new IllegalStateException(
"use_pg_ctl may not be called while server is started");
m_usePostgres = ! setting;
}
/**
* Returns a {@code Connection} to the server associated with this Node,
* using default properties appropriate for this setting.
*/
public Connection connect() throws Exception
{
return connect(new Properties());
}
/**
* Returns a {@code Connection} to the server associated with this Node,
* with suppliedProperties overriding or supplementing the ones
* that would be passed by default.
*/
public Connection connect(Map suppliedProperties)
throws Exception
{
Properties p = new Properties();
p.putAll(suppliedProperties);
return connect(p);
}
/**
* Returns a {@code Connection} to the server associated with this Node,
* with supplied properties p overriding or supplementing the ones
* that would be passed by default.
*
* By default, the connection is to the {@code postgres} database as the
* {@code postgres} user, using the password internally generated for this
* node, and with an {@code application_name} generated from a counter of
* connections for this node.
*/
public Connection connect(Properties p) throws Exception
{
String url;
String dbNameKey;
String appNameKey;
switch ( s_urlForm )
{
case URL_FORM_PGJDBC:
url = "jdbc:postgresql://localhost:" + m_port + '/';
dbNameKey = "PGDBNAME";
appNameKey = "ApplicationName";
break;
case URL_FORM_PGJDBCNG:
url = "jdbc:pgsql://localhost:" + m_port + '/';
dbNameKey = "database.name";
appNameKey = "application.name";
break;
default:
throw new UnsupportedOperationException(
"no recognized JDBC driver found to connect to the node");
}
p = (Properties)p.clone();
p.putIfAbsent(dbNameKey, "postgres");
p.putIfAbsent("user", "postgres");
p.putIfAbsent("password", m_password);
p.computeIfAbsent(appNameKey, o -> "Conn" + (m_connCount++));
if ( URL_FORM_PGJDBCNG == s_urlForm )
{
/*
* Contrary to its documentation, pgjdbc-ng does *not* accept a URL
* with the database name omitted. It is no use having it in the
* properties here; it must be appended to the URL.
*/
url += encode(p.getProperty(dbNameKey), "UTF-8");
}
Connection c = getConnection(url, p);
synchronized ( this )
{
if ( m_stopping )
{
try
{
throw new IllegalStateException(
"Node " + m_name + " is being stopped");
}
finally
{
c.close(); // add any exception as 'suppressed' to above
}
}
m_connections.put(c, null);
return c;
}
}
/**
* Sets a configuration variable on the server.
*
* This deserves a convenience method because the most familiar PostgreSQL
* syntax for SET doesn't lend itself to parameterization.
* @return a {@linkplain #q(Statement,Callable) result stream} from
* executing the statement
*/
public static Stream