Skip to content

Commit 1e6b366

Browse files
DC2-DanielKruegerDC2-DanielKrueger
committed
fix prepared statements
add config option to set the connection timeout
1 parent f0786b7 commit 1e6b366

File tree

3 files changed

+59
-74
lines changed

3 files changed

+59
-74
lines changed

‎modules/hivemq-edge-module-postgresql/src/main/java/com/hivemq/edge/adapters/postgresql/DatabaseConnection.java

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -10,19 +10,21 @@
1010

1111
public class DatabaseConnection {
1212

13+
private final @NotNull HikariConfig config;
1314
private @Nullable HikariDataSource ds;
1415

15-
public DatabaseConnection() {
16-
}
17-
18-
public void connect(final @NotNull String jdbcUrl, final @NotNull String username, final @NotNull String password) {
19-
final HikariConfig config = new HikariConfig();
16+
public DatabaseConnection(final @NotNull String jdbcUrl, final @NotNull String username, final @NotNull String password, final int connectionTimeout) {
17+
config = new HikariConfig();
2018
config.setJdbcUrl(jdbcUrl);
2119
config.setUsername(username);
2220
config.setPassword(password);
21+
config.setConnectionTimeout(connectionTimeout * 1000L);
2322
config.addDataSourceProperty("cachePrepStmts", "true");
2423
config.addDataSourceProperty("prepStmtCacheSize", "250");
2524
config.addDataSourceProperty("prepStmtCacheSqlLimit", "2048");
25+
}
26+
27+
public void connect() {
2628
this.ds = new HikariDataSource(config);
2729
}
2830

‎modules/hivemq-edge-module-postgresql/src/main/java/com/hivemq/edge/adapters/postgresql/PostgreSQLPollingProtocolAdapter.java

Lines changed: 40 additions & 68 deletions
Original file line numberDiff line numberDiff line change
@@ -30,19 +30,18 @@
3030
import com.hivemq.adapter.sdk.api.state.ProtocolAdapterState;
3131
import com.hivemq.adapter.sdk.api.tag.Tag;
3232
import com.hivemq.edge.adapters.postgresql.config.PostgreSQLAdapterConfig;
33+
import com.hivemq.edge.adapters.postgresql.config.PostgreSQLAdapterTag;
3334
import com.hivemq.edge.adapters.postgresql.config.PostgreSQLAdapterTagDefinition;
3435
import org.jetbrains.annotations.NotNull;
3536
import org.slf4j.Logger;
3637
import org.slf4j.LoggerFactory;
3738

39+
import java.sql.Connection;
3840
import java.sql.PreparedStatement;
3941
import java.sql.ResultSet;
4042
import java.sql.ResultSetMetaData;
41-
import java.sql.SQLException;
4243
import java.util.ArrayList;
43-
import java.util.HashMap;
4444
import java.util.List;
45-
import java.util.Optional;
4645

4746

4847
public class PostgreSQLPollingProtocolAdapter implements PollingProtocolAdapter {
@@ -56,12 +55,7 @@ public class PostgreSQLPollingProtocolAdapter implements PollingProtocolAdapter
5655
private final @NotNull ProtocolAdapterState protocolAdapterState;
5756
private final @NotNull String adapterId;
5857
private final @NotNull List<Tag> tags;
59-
private final @NotNull String compiledUri;
60-
private final @NotNull String username;
61-
private final @NotNull String password;
62-
private final @NotNull DatabaseConnection databaseConnection = new DatabaseConnection();
63-
private final @NotNull List<PollingContext> pollingContexts;
64-
private final @NotNull HashMap<String, PreparedStatement> tagNameToPreparedStatement = new HashMap<>();
58+
private final @NotNull DatabaseConnection databaseConnection;
6559

6660
public PostgreSQLPollingProtocolAdapter(
6761
final @NotNull ProtocolAdapterInformation adapterInformation,
@@ -71,13 +65,14 @@ public PostgreSQLPollingProtocolAdapter(
7165
this.adapterConfig = input.getConfig();
7266
this.protocolAdapterState = input.getProtocolAdapterState();
7367
this.tags = input.getTags();
74-
this.compiledUri = String.format("jdbc:postgresql://%s:%s/%s",
68+
final String compiledUri = String.format("jdbc:postgresql://%s:%s/%s",
7569
adapterConfig.getServer(),
7670
adapterConfig.getPort(),
7771
adapterConfig.getDatabase());
78-
this.username = adapterConfig.getUsername();
79-
this.password = adapterConfig.getPassword();
80-
this.pollingContexts = input.getPollingContexts();
72+
this.databaseConnection = new DatabaseConnection(compiledUri,
73+
adapterConfig.getUsername(),
74+
adapterConfig.getPassword(),
75+
adapterConfig.getConnectionTimeout());
8176
}
8277

8378
@Override
@@ -94,14 +89,7 @@ public void start(
9489
output.failStart(e, null);
9590
return;
9691
}
97-
databaseConnection.connect(compiledUri, username, password);
98-
99-
try {
100-
preparePreparedStatements(output);
101-
} catch (final SQLException e) {
102-
output.failStart(e, null);
103-
return;
104-
}
92+
databaseConnection.connect();
10593

10694
try {
10795
log.debug("Starting connection to the database instance");
@@ -119,26 +107,6 @@ public void start(
119107
}
120108
}
121109

122-
private void preparePreparedStatements(final @NotNull ProtocolAdapterStartOutput output) throws SQLException {
123-
for (final PollingContext pollingContext : pollingContexts) {
124-
final Optional<Tag> optDomainTag =
125-
tags.stream().filter(tag -> tag.getName().equals(pollingContext.getTagName())).findFirst();
126-
if (optDomainTag.isPresent()) {
127-
final PostgreSQLAdapterTagDefinition definition =
128-
(PostgreSQLAdapterTagDefinition) optDomainTag.get().getDefinition();
129-
final PreparedStatement preparedStatement =
130-
databaseConnection.getConnection().prepareStatement(definition.getQuery());
131-
tagNameToPreparedStatement.put(pollingContext.getTagName(), preparedStatement);
132-
} else {
133-
output.failStart(new IllegalStateException(
134-
"Polling for PostgreSQL protocol adapter failed because the used tag '" +
135-
pollingContext.getTagName() +
136-
"' was not found. For the polling to work the tag must be created via REST API or the UI."),
137-
null);
138-
}
139-
}
140-
}
141-
142110
@Override
143111
public void stop(
144112
final @NotNull ProtocolAdapterStopInput protocolAdapterStopInput,
@@ -164,49 +132,53 @@ public void poll(final @NotNull PollingInput pollingInput, final @NotNull Pollin
164132
tags.stream()
165133
.filter(tag -> tag.getName().equals(pollingContext.getTagName()))
166134
.findFirst()
167-
.ifPresentOrElse(def -> loadDataFromDB(pollingOutput, def),
135+
.ifPresentOrElse(tag -> loadDataFromDB(pollingOutput, (PostgreSQLAdapterTag) tag),
168136
() -> pollingOutput.fail("Polling for PostgreSQL protocol adapter failed because the used tag '" +
169137
pollingInput.getPollingContext().getTagName() +
170138
"' was not found. For the polling to work the tag must be created via REST API or the UI."));
171139
pollingOutput.finish();
172140
}
173141

174-
private void loadDataFromDB(final @NotNull PollingOutput pollingOutput, final @NotNull Tag tag) {
142+
private void loadDataFromDB(final @NotNull PollingOutput output, final @NotNull PostgreSQLAdapterTag tag) {
175143
try {
176144
log.debug("Getting tag definition");
177145
/* Get the tag definition (Query, RowLimit and Split Lines)*/
178-
final PostgreSQLAdapterTagDefinition definition = (PostgreSQLAdapterTagDefinition) tag.getDefinition();
146+
final PostgreSQLAdapterTagDefinition definition = tag.getDefinition();
179147

180148
/* Execute query and handle result */
181-
final PreparedStatement preparedStatement = tagNameToPreparedStatement.get(tag.getName());
182-
final ResultSet result = preparedStatement.executeQuery();
183-
assert result != null;
184-
final ArrayList<ObjectNode> resultObject = new ArrayList<>();
185-
final ResultSetMetaData resultSetMD = result.getMetaData();
186-
while (result.next()) {
187-
final int numColumns = resultSetMD.getColumnCount();
188-
final ObjectNode node = OBJECT_MAPPER.createObjectNode();
189-
for (int i = 1; i <= numColumns; i++) {
190-
final String column_name = resultSetMD.getColumnName(i);
191-
node.put(column_name, result.getString(column_name));
149+
try (final Connection connection = databaseConnection.getConnection()) {
150+
final PreparedStatement preparedStatement = connection.prepareStatement(tag.getDefinition().getQuery());
151+
final ResultSet result = preparedStatement.executeQuery();
152+
assert result != null;
153+
final ArrayList<ObjectNode> resultObject = new ArrayList<>();
154+
final ResultSetMetaData resultSetMD = result.getMetaData();
155+
while (result.next()) {
156+
final int numColumns = resultSetMD.getColumnCount();
157+
final ObjectNode node = OBJECT_MAPPER.createObjectNode();
158+
for (int i = 1; i <= numColumns; i++) {
159+
final String column_name = resultSetMD.getColumnName(i);
160+
node.put(column_name, result.getString(column_name));
161+
}
162+
163+
/* Publish datapoint with a single line if split is required */
164+
if (definition.getSpiltLinesInIndividualMessages()) {
165+
log.debug("Splitting lines in multiple messages");
166+
output.addDataPoint("queryResult", node);
167+
} else {
168+
resultObject.add(node);
169+
}
192170
}
193171

194-
/* Publish datapoint with a single line if split is required */
195-
if (definition.getSpiltLinesInIndividualMessages()) {
196-
log.debug("Splitting lines in multiple messages");
197-
pollingOutput.addDataPoint("queryResult", node);
198-
} else {
199-
resultObject.add(node);
172+
/* Publish datapoint with all lines if no split is required */
173+
if (!definition.getSpiltLinesInIndividualMessages()) {
174+
log.debug("Publishing all lines in a single message");
175+
output.addDataPoint("queryResult", resultObject);
200176
}
201-
}
202-
203-
/* Publish datapoint with all lines if no split is required */
204-
if (!definition.getSpiltLinesInIndividualMessages()) {
205-
log.debug("Publishing all lines in a single message");
206-
pollingOutput.addDataPoint("queryResult", resultObject);
177+
} catch (final Exception e) {
178+
output.fail(e, null);
207179
}
208180
} catch (final Exception e) {
209-
pollingOutput.fail(e, null);
181+
output.fail(e, null);
210182
}
211183
}
212184

‎modules/hivemq-edge-module-postgresql/src/main/java/com/hivemq/edge/adapters/postgresql/config/PostgreSQLAdapterConfig.java

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -89,6 +89,13 @@ public class PostgreSQLAdapterConfig implements ProtocolSpecificAdapterConfig {
8989
stringMaxLength = 1024)
9090
protected @NotNull String password;
9191

92+
@JsonProperty(value = "connectionTimeoutSeconds")
93+
@ModuleConfigField(title = "connectionTimeoutSeconds",
94+
description = "The timeout for connection establishment to the database.",
95+
numberMax = 0,
96+
defaultValue = "30")
97+
protected int connectionTimeoutSeconds = 30;
98+
9299
@JsonProperty("pollingIntervalMillis")
93100
@ModuleConfigField(title = "Polling Interval [ms]",
94101
description = "Time in millisecond that this endpoint will be polled",
@@ -99,7 +106,7 @@ public class PostgreSQLAdapterConfig implements ProtocolSpecificAdapterConfig {
99106
@JsonProperty("maxPollingErrorsBeforeRemoval")
100107
@ModuleConfigField(title = "Max. Polling Errors",
101108
description = "Max. errors polling the endpoint before the polling daemon is stopped",
102-
numberMin = 3,
109+
numberMin = -1,
103110
defaultValue = "10")
104111
private int maxPollingErrorsBeforeRemoval = 10;
105112

@@ -141,4 +148,8 @@ public int getPollingIntervalMillis() {
141148
public int getMaxPollingErrorsBeforeRemoval() {
142149
return maxPollingErrorsBeforeRemoval;
143150
}
151+
152+
public int getConnectionTimeout() {
153+
return connectionTimeoutSeconds;
154+
}
144155
}

0 commit comments

Comments
 (0)