Skip to content

Commit 95aac1f

Browse files
ronjaquenselronjaquensel
authored
feat: SQL persistence layer (#80)
* feat: SQL stores * chore: move statements to interface * test: add store tests * chore: license headers * docs: Javadoc * chore: add builder methods for stores on Dataplane * refactor: SQL stores * chore: checkstyle * chore: PR remarks
1 parent da4b72b commit 95aac1f

16 files changed

Lines changed: 945 additions & 2 deletions

‎build.gradle.kts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,9 @@ dependencies {
3434
testImplementation("org.mockito:mockito-core:5.23.0")
3535
testImplementation("org.slf4j:slf4j-simple:2.0.18")
3636
testImplementation("org.wiremock:wiremock-jetty12:3.13.2")
37+
testImplementation("org.testcontainers:testcontainers-junit-jupiter:2.0.5")
38+
testImplementation("org.testcontainers:testcontainers-postgresql:2.0.5")
39+
testImplementation("org.postgresql:postgresql:42.7.11")
3740
}
3841

3942
tasks.test {

‎src/main/java/org/eclipse/dataplane/Dataplane.java‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@
5151
import org.eclipse.dataplane.port.store.DataFlowStore;
5252
import org.eclipse.dataplane.port.store.InMemoryControlPlaneStore;
5353
import org.eclipse.dataplane.port.store.InMemoryDataFlowStore;
54+
import org.eclipse.dataplane.port.store.Stores;
5455

5556
import java.net.URI;
5657
import java.net.http.HttpClient;
@@ -69,8 +70,8 @@
6970
public class Dataplane {
7071

7172
private final ObjectMapper objectMapper = new ObjectMapper().configure(FAIL_ON_UNKNOWN_PROPERTIES, false);
72-
private final DataFlowStore dataFlowStore = new InMemoryDataFlowStore(objectMapper);
73-
private final ControlPlaneStore controlPlaneStore = new InMemoryControlPlaneStore(objectMapper);
73+
private DataFlowStore dataFlowStore = new InMemoryDataFlowStore(objectMapper);
74+
private ControlPlaneStore controlPlaneStore = new InMemoryControlPlaneStore(objectMapper);
7475
private String id;
7576
private URI endpoint;
7677
private final Set<String> transferTypes = new HashSet<>();
@@ -454,6 +455,12 @@ public Builder label(String label) {
454455
return this;
455456
}
456457

458+
public Builder stores(Stores stores) {
459+
dataplane.dataFlowStore = stores.dataFlowStore();
460+
dataplane.controlPlaneStore = stores.controlPlaneStore();
461+
return this;
462+
}
463+
457464
public Builder onPrepare(OnPrepare onPrepare) {
458465
dataplane.onPrepare = onPrepare;
459466
return this;

‎src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlow.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -243,6 +243,16 @@ public Builder callbackAddress(URI callbackAddress) {
243243
return this;
244244
}
245245

246+
public Builder suspensionReason(String suspensionReason) {
247+
dataFlow.suspensionReason = suspensionReason;
248+
return this;
249+
}
250+
251+
public Builder terminationReason(String terminationReason) {
252+
dataFlow.terminationReason = terminationReason;
253+
return this;
254+
}
255+
246256
public Builder metadata(Map<String, Object> metadata) {
247257
dataFlow.metadata = metadata;
248258
return this;

‎src/main/java/org/eclipse/dataplane/port/exception/PersistenceException.java‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
/*
2+
* Copyright (c) 2026 Fraunhofer-Gesellschaft zur Förderung der angewandten Forschung e.V.
3+
*
4+
* This program and the accompanying materials are made available under the
5+
* terms of the Apache License, Version 2.0 which is available at
6+
* https://www.apache.org/licenses/LICENSE-2.0
7+
*
8+
* SPDX-License-Identifier: Apache-2.0
9+
*
10+
* Contributors:
11+
* Fraunhofer-Gesellschaft zur Förderung der angewandten Forschung e.V. - initial API and implementation
12+
*
13+
*/
14+
15+
package org.eclipse.dataplane.port.exception;
16+
17+
/**
18+
* Indicates an error during database interactions, i.e. an error occurred persisting, reading or
19+
* deleting an entry.
20+
*/
21+
public class PersistenceException extends RuntimeException {
22+
23+
public PersistenceException(String message) {
24+
super(message);
25+
}
26+
27+
public PersistenceException(String message, Throwable cause) {
28+
super(message, cause);
29+
}
30+
}

‎src/main/java/org/eclipse/dataplane/port/store/Stores.java‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
/*
2+
* Copyright (c) 2026 Fraunhofer-Gesellschaft zur Förderung der angewandten Forschung e.V.
3+
*
4+
* This program and the accompanying materials are made available under the
5+
* terms of the Apache License, Version 2.0 which is available at
6+
* https://www.apache.org/licenses/LICENSE-2.0
7+
*
8+
* SPDX-License-Identifier: Apache-2.0
9+
*
10+
* Contributors:
11+
* Fraunhofer-Gesellschaft zur Förderung der angewandten Forschung e.V. - initial API and implementation
12+
*
13+
*/
14+
15+
package org.eclipse.dataplane.port.store;
16+
17+
/**
18+
* Data class that bundles the stores used by the dataplane.
19+
*
20+
* @param dataFlowStore store for data flows
21+
* @param controlPlaneStore store for control planes
22+
*/
23+
public record Stores(DataFlowStore dataFlowStore, ControlPlaneStore controlPlaneStore) {
24+
}

‎src/main/java/org/eclipse/dataplane/port/store/sql/AbstractSqlStore.java‎

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,92 @@
1+
/*
2+
* Copyright (c) 2026 Fraunhofer-Gesellschaft zur Förderung der angewandten Forschung e.V.
3+
*
4+
* This program and the accompanying materials are made available under the
5+
* terms of the Apache License, Version 2.0 which is available at
6+
* https://www.apache.org/licenses/LICENSE-2.0
7+
*
8+
* SPDX-License-Identifier: Apache-2.0
9+
*
10+
* Contributors:
11+
* Fraunhofer-Gesellschaft zur Förderung der angewandten Forschung e.V. - initial API and implementation
12+
*
13+
*/
14+
15+
package org.eclipse.dataplane.port.store.sql;
16+
17+
import com.fasterxml.jackson.core.JsonProcessingException;
18+
import com.fasterxml.jackson.core.type.TypeReference;
19+
import com.fasterxml.jackson.databind.JavaType;
20+
import com.fasterxml.jackson.databind.ObjectMapper;
21+
import org.eclipse.dataplane.port.exception.PersistenceException;
22+
23+
import java.sql.Connection;
24+
import java.sql.DriverManager;
25+
26+
/**
27+
* Base class for SQL-based store implementations that provides methods for common functionality
28+
* like connection handling and JSON parsing.
29+
*/
30+
public abstract class AbstractSqlStore {
31+
32+
protected ObjectMapper objectMapper;
33+
34+
private final String databaseUrl;
35+
private final String databaseUsername;
36+
private final String databasePassword;
37+
38+
public AbstractSqlStore(ObjectMapper objectMapper, String databaseUrl, String databaseUsername, String databasePassword) {
39+
this.objectMapper = objectMapper;
40+
this.databaseUrl = databaseUrl;
41+
this.databaseUsername = databaseUsername;
42+
this.databasePassword = databasePassword;
43+
}
44+
45+
protected Connection getConnection() {
46+
try {
47+
return DriverManager.getConnection(databaseUrl, databaseUsername, databasePassword);
48+
} catch (Exception e) {
49+
throw new PersistenceException("Failed to connect to database.", e);
50+
}
51+
}
52+
53+
protected void closeConnection(Connection connection) {
54+
try {
55+
connection.close();
56+
} catch (Exception e) {
57+
throw new PersistenceException("Failed to commit transaction.", e);
58+
}
59+
}
60+
61+
protected String toJson(Object object) {
62+
if (object == null) {
63+
return null;
64+
}
65+
66+
try {
67+
return object instanceof String ? object.toString() : objectMapper.writeValueAsString(object);
68+
} catch (JsonProcessingException e) {
69+
throw new PersistenceException("Failed to convert object to JSON.", e);
70+
}
71+
}
72+
73+
protected <T> T fromJson(String json, Class<T> type) {
74+
return fromJson(json, objectMapper.getTypeFactory().constructType(type));
75+
}
76+
77+
protected <T> T fromJson(String json, TypeReference<T> type) {
78+
return fromJson(json, objectMapper.getTypeFactory().constructType(type));
79+
}
80+
81+
protected <T> T fromJson(String json, JavaType type) {
82+
if (json == null) {
83+
return null;
84+
}
85+
86+
try {
87+
return objectMapper.readValue(json, type);
88+
} catch (JsonProcessingException e) {
89+
throw new PersistenceException("Failed to convert JSON to object.", e);
90+
}
91+
}
92+
}

‎src/main/java/org/eclipse/dataplane/port/store/sql/PostgresControlPlaneStore.java‎

Lines changed: 130 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,130 @@
1+
/*
2+
* Copyright (c) 2026 Fraunhofer-Gesellschaft zur Förderung der angewandten Forschung e.V.
3+
*
4+
* This program and the accompanying materials are made available under the
5+
* terms of the Apache License, Version 2.0 which is available at
6+
* https://www.apache.org/licenses/LICENSE-2.0
7+
*
8+
* SPDX-License-Identifier: Apache-2.0
9+
*
10+
* Contributors:
11+
* Fraunhofer-Gesellschaft zur Förderung der angewandten Forschung e.V. - initial API and implementation
12+
*
13+
*/
14+
15+
package org.eclipse.dataplane.port.store.sql;
16+
17+
import com.fasterxml.jackson.databind.ObjectMapper;
18+
import org.eclipse.dataplane.domain.Result;
19+
import org.eclipse.dataplane.domain.controlplane.ControlPlane;
20+
import org.eclipse.dataplane.domain.registration.AuthorizationProfile;
21+
import org.eclipse.dataplane.port.exception.PersistenceException;
22+
import org.eclipse.dataplane.port.exception.ResourceNotFoundException;
23+
import org.eclipse.dataplane.port.store.ControlPlaneStore;
24+
25+
import java.net.URI;
26+
27+
import static java.lang.String.format;
28+
29+
public class PostgresControlPlaneStore extends AbstractSqlStore implements ControlPlaneStore {
30+
31+
public PostgresControlPlaneStore(ObjectMapper objectMapper, String databaseUrl, String databaseUsername, String databasePassword) {
32+
super(objectMapper, databaseUrl, databaseUsername, databasePassword);
33+
}
34+
35+
@Override
36+
public Result<Void> save(ControlPlane controlPlane) {
37+
var connection = getConnection();
38+
39+
try (var statement = connection.prepareStatement(upsertControlPlaneTemplate())) {
40+
statement.setString(1, controlPlane.getId());
41+
statement.setString(2, controlPlane.getEndpoint().toString());
42+
statement.setString(3, toJson(controlPlane.getAuthorization()));
43+
44+
statement.executeUpdate();
45+
return Result.success();
46+
} catch (Exception e) {
47+
return Result.failure(new PersistenceException(format("Failed to persist ControlPlane with id %s.", controlPlane.getId()), e));
48+
} finally {
49+
closeConnection(connection);
50+
}
51+
}
52+
53+
@Override
54+
public Result<ControlPlane> findById(String controlplaneId) {
55+
var connection = getConnection();
56+
57+
try (var statement = connection.prepareStatement(findControlPlaneByIdTemplate())) {
58+
statement.setString(1, controlplaneId);
59+
var resultSet = statement.executeQuery();
60+
61+
if (!resultSet.next()) {
62+
return Result.failure(new ResourceNotFoundException(format("ControlPlane with id %s not found.", controlplaneId)));
63+
}
64+
65+
var controlplane = ControlPlane.newInstance()
66+
.id(controlplaneId)
67+
.endpoint(URI.create(resultSet.getString("endpoint")))
68+
.authorization(fromJson(resultSet.getString("auth"), AuthorizationProfile.class))
69+
.build();
70+
return Result.success(controlplane);
71+
} catch (Exception e) {
72+
return Result.failure(new PersistenceException(format("Failed to read ControlPlane with id %s.", controlplaneId), e));
73+
} finally {
74+
closeConnection(connection);
75+
}
76+
}
77+
78+
@Override
79+
public Result<Void> delete(String id) {
80+
var connection = getConnection();
81+
82+
try (var statement = connection.prepareStatement(deleteControlPlaneByIdTemplate())) {
83+
statement.setString(1, id);
84+
var rows = statement.executeUpdate();
85+
if (rows < 1) {
86+
return Result.failure(new ResourceNotFoundException(format("ControlPlane with id %s not found.", id)));
87+
}
88+
return Result.success();
89+
} catch (Exception e) {
90+
return Result.failure(new PersistenceException(format("Failed to delete ControlPlane with id %s.", id), e));
91+
} finally {
92+
closeConnection(connection);
93+
}
94+
}
95+
96+
@Override
97+
public boolean exists(String controlplaneId) {
98+
var connection = getConnection();
99+
100+
try (var statement = connection.prepareStatement(countControlPlaneByIdTemplate())) {
101+
statement.setString(1, controlplaneId);
102+
var resultSet = statement.executeQuery();
103+
resultSet.next();
104+
return resultSet.getInt(1) > 0;
105+
} catch (Exception e) {
106+
throw new PersistenceException(format("Failed to check for existence of ControlPlane with id %s.", controlplaneId), e);
107+
} finally {
108+
closeConnection(connection);
109+
}
110+
}
111+
112+
private String upsertControlPlaneTemplate() {
113+
return "INSERT INTO control_planes (id, endpoint, auth) VALUES (?, ?, ?::json)" +
114+
" ON CONFLICT (id) DO UPDATE SET" +
115+
" endpoint = EXCLUDED.endpoint," +
116+
" auth = EXCLUDED.auth";
117+
}
118+
119+
private String findControlPlaneByIdTemplate() {
120+
return "SELECT * FROM control_planes WHERE id = ?";
121+
}
122+
123+
private String deleteControlPlaneByIdTemplate() {
124+
return "DELETE FROM control_planes WHERE id = ?";
125+
}
126+
127+
private String countControlPlaneByIdTemplate() {
128+
return "SELECT COUNT(*) FROM control_planes WHERE id = ?";
129+
}
130+
}

0 commit comments

Comments
 (0)