Skip to content

Postgres plugin: supported configurable host:port #120

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lvca merged 2 commits into from
Oct 10, 2021
Merged

Postgres plugin: supported configurable host:port #120

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5a11f86
Postgres plugin: supported configurable host:port
lvca Oct 10, 2021
95a6c9e
Postgres: disabled debug
lvca Oct 10, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions engine/src/main/java/com/arcadedb/GlobalConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -269,6 +269,11 @@ public Object call(final Object value) {
1000),

// GREMLIN

// POSTGRES
POSTGRES_PORT("arcadedb.postgres.port", "TCP/IP port number used for incoming connections for Postgres plugin. Default is 5432", Integer.class, 5432),

POSTGRES_HOST("arcadedb.postgres.host", "TCP/IP host name used for incoming connections for Postgres plugin. Default is '0.0.0.0'", String.class, "0.0.0.0"),
;

/**
Expand Down
29 changes: 18 additions & 11 deletions postgresw/src/main/java/com/arcadedb/postgres/PostgresNetworkExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -203,7 +203,7 @@ public void run() {

private void syncCommand() {
if (DEBUG)
LogManager.instance().log(this, Level.INFO, "PSQL: sync");
LogManager.instance().log(this, Level.INFO, "PSQL: sync (thread=%s)", null, Thread.currentThread().getId());

if (!explicitTransactionStarted) {
if (errorInTransaction)
Expand All @@ -221,10 +221,12 @@ private void closeCommand() throws IOException {
final byte closeType = channel.readByte();
final String prepStatementOrPortal = readString();

getPortal(prepStatementOrPortal, true);
if (closeType == 'P')
getPortal(prepStatementOrPortal, true);

if (DEBUG)
LogManager.instance().log(this, Level.INFO, "PSQL: close '%s' type=%s", null, prepStatementOrPortal, (char) closeType);
LogManager.instance()
.log(this, Level.INFO, "PSQL: close '%s' type=%s (thread=%s)", null, prepStatementOrPortal, (char) closeType, Thread.currentThread().getId());

writeMessage("close complete", null, '3', 4);
}
Expand All @@ -234,7 +236,7 @@ private void describeCommand() throws IOException {
final String portalName = readString();

if (DEBUG)
LogManager.instance().log(this, Level.INFO, "PSQL: describe '%s' type=%s", null, portalName, (char) type);
LogManager.instance().log(this, Level.INFO, "PSQL: describe '%s' type=%s (thread=%s)", null, portalName, (char) type, Thread.currentThread().getId());

final PostgresPortal portal = getPortal(portalName, false);
if (portal == null) {
Expand Down Expand Up @@ -277,7 +279,8 @@ private void executeCommand() {
}

if (DEBUG)
LogManager.instance().log(this, Level.INFO, "PSQL: execute (portal=%s) (limit=%d)-> %s", null, portalName, limit, portal);
LogManager.instance()
.log(this, Level.INFO, "PSQL: execute (portal=%s) (limit=%d)-> %s (thread=%s)", null, portalName, limit, portal, Thread.currentThread().getId());

if (portal.ignoreExecution)
writeMessage("empty query response", null, 'I', 4);
Expand Down Expand Up @@ -322,7 +325,7 @@ private void queryCommand() {
queryText = queryText.substring(0, queryText.length() - 1);

if (DEBUG)
LogManager.instance().log(this, Level.INFO, "PSQL: query -> %s", null, queryText);
LogManager.instance().log(this, Level.INFO, "PSQL: query -> %s (thread=%s)", null, queryText, Thread.currentThread().getId());

if (queryText.isEmpty()) {

Expand Down Expand Up @@ -481,7 +484,8 @@ private void writeDataRows(final List<Result> resultSet, final Map<String, Postg
channel.flush();

if (DEBUG)
LogManager.instance().log(this, Level.INFO, "PSQL:-> %d row data (%s)", null, resultSet.size(), FileUtils.getSizeAsString(bufferData.limit()));
LogManager.instance().log(this, Level.INFO, "PSQL:-> %d row data (%s) (thread=%s)", null, resultSet.size(), FileUtils.getSizeAsString(bufferData.limit()),
Thread.currentThread().getId());
}

private void bindCommand() {
Expand All @@ -497,7 +501,8 @@ private void bindCommand() {
}

if (DEBUG)
LogManager.instance().log(this, Level.INFO, "PSQL: bind (portal=%s) -> %s", null, portalName, sourcePreparedStatement);
LogManager.instance()
.log(this, Level.INFO, "PSQL: bind (portal=%s) -> %s (thread=%s)", null, portalName, sourcePreparedStatement, Thread.currentThread().getId());

final int paramFormatCount = channel.readShort();
if (paramFormatCount > 0) {
Expand Down Expand Up @@ -557,7 +562,8 @@ private void parseCommand() {
}

if (DEBUG)
LogManager.instance().log(this, Level.INFO, "PSQL: parse (portal=%s) -> %s (params=%d)", null, portalName, portal.query, paramCount);
LogManager.instance().log(this, Level.INFO, "PSQL: parse (portal=%s) -> %s (params=%d) (thread=%s)", null, portalName, portal.query, paramCount,
Thread.currentThread().getId());

final String upperCaseText = portal.query.toUpperCase();
if (upperCaseText.startsWith("SET ")) {
Expand Down Expand Up @@ -805,13 +811,14 @@ private void writeMessage(final String messageName, final WriteMessageCallback c
channel.flush();

if (DEBUG)
LogManager.instance().log(this, Level.INFO, "PSQL:-> %s (%s - %s)", null, messageName, messageCode, FileUtils.getSizeAsString(length));
LogManager.instance().log(this, Level.INFO, "PSQL:-> %s (%s - %s) (thread=%s)", null, messageName, messageCode, FileUtils.getSizeAsString(length),
Thread.currentThread().getId());

} catch (IOException e) {
if (database.isTransactionActive())
errorInTransaction = true;

throw new PostgresProtocolException("Error on sending " + messageName + " message", e);
throw new PostgresProtocolException("Error on sending '" + messageName + "' message", e);
}
}

Expand Down
8 changes: 4 additions & 4 deletions postgresw/src/main/java/com/arcadedb/postgres/PostgresNetworkListener.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,14 +41,14 @@ public interface ClientConnected {
private int port;
private ClientConnected callback;

public PostgresNetworkListener(final ArcadeDBServer server, final ServerSocketFactory iSocketFactory, final String iHostName, final String iHostPortRange) {
super(server.getServerName() + " PostgresW listening at " + iHostName + ":" + iHostPortRange);
public PostgresNetworkListener(final ArcadeDBServer server, final ServerSocketFactory iSocketFactory, final String hostName, final String hostPortRange) {
super(server.getServerName() + " PostgresW listening at " + hostName + ":" + hostPortRange);

this.server = server;
this.hostName = iHostName;
this.hostName = hostName;
this.socketFactory = iSocketFactory == null ? ServerSocketFactory.getDefault() : iSocketFactory;

listen(iHostName, iHostPortRange);
listen(hostName, hostPortRange);

start();
}
Expand Down
14 changes: 9 additions & 5 deletions postgresw/src/main/java/com/arcadedb/postgres/PostgresProtocolPlugin.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,27 +16,31 @@
package com.arcadedb.postgres;

import com.arcadedb.ContextConfiguration;
import com.arcadedb.GlobalConfiguration;
import com.arcadedb.server.ArcadeDBServer;
import com.arcadedb.server.ServerPlugin;
import com.arcadedb.server.ha.network.DefaultServerSocketFactory;
import com.arcadedb.server.http.HttpServer;
import io.undertow.server.handlers.PathHandler;

public class PostgresProtocolPlugin implements ServerPlugin {
private ArcadeDBServer server;
private ContextConfiguration configuration;
private final static int DEF_PORT = 5432;
private PostgresNetworkListener listener;
private ArcadeDBServer server;
private ContextConfiguration configuration;
private PostgresNetworkListener listener;
private String host;
private int port;

@Override
public void configure(final ArcadeDBServer arcadeDBServer, final ContextConfiguration configuration) {
this.server = arcadeDBServer;
this.configuration = configuration;
this.host = configuration.getValueAsString(GlobalConfiguration.POSTGRES_HOST);
this.port = configuration.getValueAsInteger(GlobalConfiguration.POSTGRES_PORT);
}

@Override
public void startService() {
listener = new PostgresNetworkListener(server, new DefaultServerSocketFactory(), "localhost", "" + DEF_PORT);
listener = new PostgresNetworkListener(server, new DefaultServerSocketFactory(), host, "" + port);
}

@Override
Expand Down
Loading