Skip to content

Commit 8f9e2e4

Browse files
ndr-brtndr-brt
committed
build: align with spec 1.0-RC3
1 parent 24fa08f commit 8f9e2e4

19 files changed

Lines changed: 136 additions & 134 deletions

File tree

‎dataplane-sdk-core/src/main/java/org/eclipse/dataplane/Dataplane.java‎

Lines changed: 50 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -75,7 +75,7 @@ public class Dataplane {
7575
private ControlPlaneStore controlPlaneStore = new InMemoryControlPlaneStore(objectMapper);
7676
private String id;
7777
private URI endpoint;
78-
private final Set<String> transferTypes = new HashSet<>();
78+
private final Set<String> profiles = new HashSet<>();
7979
private final Set<String> labels = new HashSet<>();
8080

8181
private OnPrepare onPrepare = dataFlow -> Result.failure(new UnsupportedOperationException("onPrepare is not implemented"));
@@ -106,67 +106,59 @@ public Result<DataFlowStatusResponseMessage> status(String dataFlowId) {
106106
.map(f -> new DataFlowStatusResponseMessage(f.getId(), f.getState().name()));
107107
}
108108

109-
private Result<Void> checkControlPlane(String controlplaneId) {
110-
if (controlPlaneStore.exists(controlplaneId)) {
111-
return Result.success();
112-
}
113-
return Result.failure(new ControlPlaneNotRegistered(controlplaneId));
114-
}
115-
116109
public Result<DataFlowStatusMessage> prepare(String controlplaneId, DataFlowPrepareMessage message) {
117-
var initialDataFlow = DataFlow.newInstance()
118-
.id(message.processId())
119-
.state(DataFlow.State.INITIATING)
120-
.labels(message.labels())
121-
.metadata(message.metadata())
122-
.callbackAddress(message.callbackAddress())
123-
.transferType(message.transferType())
124-
.datasetId(message.datasetId())
125-
.agreementId(message.agreementId())
126-
.participantId(message.participantId())
127-
.counterPartyId(message.counterPartyId())
128-
.dataspaceContext(message.dataspaceContext())
129-
.controlplaneId(controlplaneId)
130-
.type(DataFlow.Type.CONSUMER)
131-
.build();
132-
133-
return checkControlPlane(controlplaneId)
134-
.compose(v -> onPrepare.action(initialDataFlow))
110+
return getControlPlane(controlplaneId)
111+
.map(controlPlane -> DataFlow.newInstance()
112+
.id(message.processId())
113+
.state(DataFlow.State.INITIATING)
114+
.labels(message.labels())
115+
.metadata(message.metadata())
116+
.callbackAddress(controlPlane.getEndpoint())
117+
.profile(message.profile())
118+
.datasetId(message.datasetId())
119+
.agreementId(message.agreementId())
120+
.participantId(message.participantId())
121+
.counterPartyId(message.counterPartyId())
122+
.dataspaceContext(message.dataspaceContext())
123+
.controlplaneId(controlplaneId)
124+
.type(DataFlow.Type.CONSUMER)
125+
.build()
126+
)
127+
.compose(initialDataFlow -> onPrepare.action(initialDataFlow))
135128
.compose(dataFlow -> {
136129
if (dataFlow.isInitiating()) {
137130
dataFlow.transitionToPrepared();
138131
}
139132

140133
DataFlowStatusMessage response;
141134
if (dataFlow.isPrepared() && dataFlow.isPush()) {
142-
response = new DataFlowStatusMessage(dataFlow.getId(), initialDataFlow.getState().name(), dataFlow.getDataAddress(), null);
135+
response = new DataFlowStatusMessage(dataFlow.getId(), dataFlow.getState().name(), dataFlow.getDataAddress(), null);
143136
} else {
144-
response = new DataFlowStatusMessage(dataFlow.getId(), initialDataFlow.getState().name(), null, null);
137+
response = new DataFlowStatusMessage(dataFlow.getId(), dataFlow.getState().name(), null, null);
145138
}
146139

147140
return save(dataFlow).map(it -> response);
148141
});
149142
}
150143

151-
152144
public Result<DataFlowStatusMessage> start(String controlplaneId, DataFlowStartMessage message) {
153-
var initialDataFlow = DataFlow.newInstance()
154-
.id(message.processId())
155-
.state(DataFlow.State.INITIATING)
156-
.dataAddress(message.dataAddress())
157-
.callbackAddress(message.callbackAddress())
158-
.transferType(message.transferType())
159-
.datasetId(message.datasetId())
160-
.agreementId(message.agreementId())
161-
.participantId(message.participantId())
162-
.counterPartyId(message.counterPartyId())
163-
.dataspaceContext(message.dataspaceContext())
164-
.controlplaneId(controlplaneId)
165-
.type(DataFlow.Type.PROVIDER)
166-
.build();
167-
168-
return checkControlPlane(controlplaneId)
169-
.compose(v -> onStart.action(initialDataFlow))
145+
return getControlPlane(controlplaneId)
146+
.map(controlPlane -> DataFlow.newInstance()
147+
.id(message.processId())
148+
.state(DataFlow.State.INITIATING)
149+
.dataAddress(message.dataAddress())
150+
.callbackAddress(controlPlane.getEndpoint())
151+
.profile(message.profile())
152+
.datasetId(message.datasetId())
153+
.agreementId(message.agreementId())
154+
.participantId(message.participantId())
155+
.counterPartyId(message.counterPartyId())
156+
.dataspaceContext(message.dataspaceContext())
157+
.controlplaneId(controlplaneId)
158+
.type(DataFlow.Type.PROVIDER)
159+
.build()
160+
)
161+
.compose(initialDataFlow -> onStart.action(initialDataFlow))
170162
.compose(dataFlow -> {
171163
if (dataFlow.isInitiating()) {
172164
dataFlow.transitionToStarted();
@@ -325,7 +317,7 @@ public Result<String> extractControlplaneId(String authorizationHeader) {
325317

326318
public Result<Void> registerOn(String controlPlaneEndpoint) {
327319

328-
var message = new DataPlaneRegistrationMessage(id, endpoint, transferTypes, labels);
320+
var message = new DataPlaneRegistrationMessage(id, endpoint, profiles, labels);
329321

330322
return toJson(message)
331323
.map(body -> HttpRequest.newBuilder()
@@ -344,6 +336,14 @@ public Result<Void> registerOn(String controlPlaneEndpoint) {
344336
});
345337
}
346338

339+
private Result<ControlPlane> getControlPlane(String controlplaneId) {
340+
var controlPlaneById = controlPlaneStore.findById(controlplaneId);
341+
if (controlPlaneById.failed()) {
342+
return Result.failure(new ControlPlaneNotRegistered(controlplaneId));
343+
}
344+
return controlPlaneById;
345+
}
346+
347347
private DataAddress getDataAddressForResume(DataFlow dataFlow) {
348348
if (dataFlow.isPull() && dataFlow.getType() == DataFlow.Type.PROVIDER) {
349349
return dataFlow.getDataAddress();
@@ -363,7 +363,7 @@ private Result<Void> notifyControlPlane(String action, DataFlow dataFlow, Object
363363
.header("content-type", "application/json")
364364
.POST(HttpRequest.BodyPublishers.ofString(body));
365365

366-
controlPlaneStore.findById(dataFlow.getControlplaneId())
366+
getControlPlane(dataFlow.getControlplaneId())
367367
.compose(controlPlane -> {
368368
var authorizationProfile = controlPlane.getAuthorization();
369369
if (authorizationProfile != null) {
@@ -445,8 +445,8 @@ public Builder endpoint(URI endpoint) {
445445
return this;
446446
}
447447

448-
public Builder transferType(String transferType) {
449-
dataplane.transferTypes.add(transferType);
448+
public Builder profile(String profile) {
449+
dataplane.profiles.add(profile);
450450
return this;
451451
}
452452

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

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@ public class DataFlow {
2626

2727
private String id;
2828
private State state;
29-
private String transferType;
29+
private String profile;
3030
private String datasetId;
3131
private String agreementId;
3232
private String participantId;
@@ -61,8 +61,8 @@ public URI getCallbackAddress() {
6161
return callbackAddress;
6262
}
6363

64-
public String getTransferType() {
65-
return transferType;
64+
public String getProfile() {
65+
return profile;
6666
}
6767

6868
public String getDatasetId() {
@@ -132,11 +132,11 @@ public void transitionToTerminated(String reason) {
132132
}
133133

134134
public boolean isPush() {
135-
return transferTypeLastToken().equalsIgnoreCase("push");
135+
return profileLastToken().equalsIgnoreCase("push");
136136
}
137137

138138
public boolean isPull() {
139-
return transferTypeLastToken().equalsIgnoreCase("pull");
139+
return profileLastToken().equalsIgnoreCase("pull");
140140
}
141141

142142
public boolean isInitiating() {
@@ -167,8 +167,8 @@ public Type getType() {
167167
return type;
168168
}
169169

170-
private String transferTypeLastToken() {
171-
return transferType.substring(transferType.lastIndexOf('-') + 1);
170+
private String profileLastToken() {
171+
return profile.substring(profile.lastIndexOf('-') + 1);
172172
}
173173

174174
public enum Type {
@@ -202,8 +202,8 @@ public Builder state(State state) {
202202
return this;
203203
}
204204

205-
public Builder transferType(String transferType) {
206-
dataFlow.transferType = transferType;
205+
public Builder profile(String profile) {
206+
dataFlow.profile = profile;
207207
return this;
208208
}
209209

‎dataplane-sdk-core/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowPrepareMessage.java‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,6 @@
1414

1515
package org.eclipse.dataplane.domain.dataflow;
1616

17-
import java.net.URI;
1817
import java.util.List;
1918
import java.util.Map;
2019

@@ -26,8 +25,7 @@ public record DataFlowPrepareMessage(
2625
String processId,
2726
String agreementId,
2827
String datasetId,
29-
URI callbackAddress,
30-
String transferType,
28+
String profile,
3129
Map<String, Object> claims,
3230
List<String> labels,
3331
Map<String, Object> metadata

‎dataplane-sdk-core/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowStartMessage.java‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,6 @@
1616

1717
import org.eclipse.dataplane.domain.DataAddress;
1818

19-
import java.net.URI;
2019
import java.util.List;
2120
import java.util.Map;
2221

@@ -28,8 +27,7 @@ public record DataFlowStartMessage(
2827
String processId,
2928
String agreementId,
3029
String datasetId,
31-
URI callbackAddress,
32-
String transferType,
30+
String profile,
3331
DataAddress dataAddress,
3432
Map<String, Object> claims,
3533
List<String> labels,

‎dataplane-sdk-core/src/main/java/org/eclipse/dataplane/domain/registration/DataPlaneRegistrationMessage.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@
2020
public record DataPlaneRegistrationMessage(
2121
String dataplaneId,
2222
URI endpoint,
23-
Set<String> transferTypes,
23+
Set<String> profiles,
2424
Set<String> labels
2525
// TODO: authorization
2626
) {

‎dataplane-sdk-core/src/main/java/org/eclipse/dataplane/port/store/ControlPlaneStore.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -56,5 +56,6 @@ public interface ControlPlaneStore {
5656
* @param controlplaneId the id of the ControlPlane
5757
* @return true, if the ControlPlane exists in the store, false otherwise
5858
*/
59+
@Deprecated(since = "1.0.0")
5960
boolean exists(String controlplaneId);
6061
}

‎dataplane-sdk-core/src/testFixtures/java/org/eclipse/dataplane/store/DataFlowStoreTestBase.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -92,7 +92,7 @@ private DataFlow dataFlow(String id) {
9292
return DataFlow.newInstance()
9393
.id(id)
9494
.state(DataFlow.State.INITIATING)
95-
.transferType("HTTP-PUSH")
95+
.profile("HTTP-PUSH")
9696
.datasetId("dataset")
9797
.agreementId("agreement")
9898
.participantId("participant")

‎dataplane-sdk-postgresql/src/main/java/org/eclipse/dataplane/store/postgresql/PostgresDataFlowStore.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,7 @@ public Result<Void> save(DataFlow dataFlow) {
4040

4141
try (var statement = connection.prepareStatement(upsertDataFlowTemplate())) {
4242
statement.setString(1, dataFlow.getId());
43-
statement.setString(2, dataFlow.getTransferType());
43+
statement.setString(2, dataFlow.getProfile());
4444
statement.setString(3, dataFlow.getType().name());
4545
statement.setString(4, dataFlow.getState().name());
4646
statement.setString(5, dataFlow.getDatasetId());
@@ -80,7 +80,7 @@ public Result<DataFlow> findById(String flowId) {
8080
var dataFlow = DataFlow.newInstance()
8181
.id(flowId)
8282
.state(DataFlow.State.valueOf(resultSet.getString("state")))
83-
.transferType(resultSet.getString("transfer_type"))
83+
.profile(resultSet.getString("transfer_type"))
8484
.datasetId(resultSet.getString("dataset_id"))
8585
.agreementId(resultSet.getString("agreement_id"))
8686
.participantId(resultSet.getString("participant_id"))

‎e2e-tests/src/test/java/org/eclipse/dataplane/DataplaneTest.java‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -77,7 +77,7 @@ void shouldFail_whenDataFlowDoesNotExist() {
7777
@Test
7878
void shouldReturnFailedFuture_whenControlPlaneIsNotAvailable() {
7979
var dataplane = Dataplane.newInstance().onPrepare(Result::success).build();
80-
dataplane.registerControlPlane(new ControlPlaneRegistrationMessage("controlplaneId", URI.create("http://localhost/any")));
80+
dataplane.registerControlPlane(new ControlPlaneRegistrationMessage("controlplaneId", URI.create(controlPlane.baseUrl())));
8181
dataplane.prepare("controlplaneId", createPrepareMessage());
8282
controlPlane.stop();
8383

@@ -92,7 +92,7 @@ void shouldReturnFailedFuture_whenControlPlaneRespondWithError() {
9292
controlPlane.stubFor(post(anyUrl()).willReturn(aResponse().withStatus(500)));
9393

9494
var dataplane = Dataplane.newInstance().onPrepare(Result::success).build();
95-
dataplane.registerControlPlane(new ControlPlaneRegistrationMessage("controlplaneId", URI.create("http://localhost/any")));
95+
dataplane.registerControlPlane(new ControlPlaneRegistrationMessage("controlplaneId", URI.create(controlPlane.baseUrl())));
9696
dataplane.prepare("controlplaneId", createPrepareMessage());
9797

9898
var result = dataplane.notifyCompleted("dataFlowId");
@@ -106,7 +106,7 @@ void shouldReturnFailedFuture_whenControlPlaneRespondWithError() {
106106
void shouldTransitionToCompleted_whenControlPlaneRespondCorrectly() {
107107
controlPlane.stubFor(post(anyUrl()).willReturn(aResponse().withStatus(200)));
108108
var dataplane = Dataplane.newInstance().onPrepare(Result::success).build();
109-
dataplane.registerControlPlane(new ControlPlaneRegistrationMessage("controlplaneId", URI.create("http://localhost/any")));
109+
dataplane.registerControlPlane(new ControlPlaneRegistrationMessage("controlplaneId", URI.create(controlPlane.baseUrl())));
110110
dataplane.prepare("controlplaneId", createPrepareMessage());
111111

112112
var result = dataplane.notifyCompleted("dataFlowId");
@@ -132,7 +132,7 @@ void shouldFail_whenDataFlowDoesNotExist() {
132132
void shouldSendDataFlowStatusMessage_whenDataFlowIsErrored() {
133133
controlPlane.stubFor(post(anyUrl()).willReturn(aResponse().withStatus(200)));
134134
var dataplane = Dataplane.newInstance().id("dataplane-id").onPrepare(Result::success).build();
135-
dataplane.registerControlPlane(new ControlPlaneRegistrationMessage("controlplaneId", URI.create("http://localhost/any")));
135+
dataplane.registerControlPlane(new ControlPlaneRegistrationMessage("controlplaneId", URI.create(controlPlane.baseUrl())));
136136
dataplane.prepare("controlplaneId", createPrepareMessage());
137137

138138
var result = dataplane.notifyErrored("dataFlowId", new RuntimeException("some-error"));
@@ -161,7 +161,7 @@ void shouldRegisterOnTheControlPlane() {
161161
var dataplane = Dataplane.newInstance()
162162
.id("dataplane-id")
163163
.endpoint(URI.create("http://localhost/dataplane"))
164-
.transferType("SupportedTransferType-PUSH")
164+
.profile("SupportedProfile-PUSH")
165165
.label("label-one").label("label-two")
166166
.build();
167167

@@ -171,7 +171,7 @@ void shouldRegisterOnTheControlPlane() {
171171
controlPlane.verify(putRequestedFor(urlPathEqualTo("/dataplanes"))
172172
.withRequestBody(and(
173173
matchingJsonPath("endpoint", equalTo("http://localhost/dataplane")),
174-
matchingJsonPath("transferTypes[0]", equalTo("SupportedTransferType-PUSH")),
174+
matchingJsonPath("profiles[0]", equalTo("SupportedProfile-PUSH")),
175175
matchingJsonPath("labels.size()", equalTo("2"))
176176
))
177177
);
@@ -184,7 +184,7 @@ void shouldFail_whenStatusIsNot200() {
184184
var dataplane = Dataplane.newInstance()
185185
.id("dataplane-id")
186186
.endpoint(URI.create("http://localhost/dataplane"))
187-
.transferType("SupportedTransferType-PUSH")
187+
.profile("SupportedProfile-PUSH")
188188
.label("label-one").label("label-two")
189189
.build();
190190

@@ -196,6 +196,6 @@ void shouldFail_whenStatusIsNot200() {
196196
}
197197

198198
private DataFlowPrepareMessage createPrepareMessage() {
199-
return MessageFactory.createPrepareMessage("dataFlowId", URI.create(controlPlane.baseUrl()), "Something-PUSH");
199+
return MessageFactory.createPrepareMessage("dataFlowId", "Something-PUSH");
200200
}
201201
}

‎e2e-tests/src/test/java/org/eclipse/dataplane/MessageFactory.java‎

Lines changed: 8 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -19,26 +19,24 @@
1919
import org.eclipse.dataplane.domain.dataflow.DataFlowStartMessage;
2020
import org.jspecify.annotations.NonNull;
2121

22-
import java.net.URI;
23-
2422
import static java.util.Collections.emptyList;
2523
import static java.util.Collections.emptyMap;
2624

2725
public interface MessageFactory {
2826

29-
static @NonNull DataFlowPrepareMessage createPrepareMessage(String consumerProcessId, URI callbackAddress, String transferType) {
27+
static @NonNull DataFlowPrepareMessage createPrepareMessage(String consumerProcessId, String profile) {
3028
return new DataFlowPrepareMessage("theMessageId", "theParticipantId", "theCounterPartyId",
31-
"theDataspaceContext", consumerProcessId, "theAgreementId", "theDatasetId", callbackAddress,
32-
transferType, emptyMap(), emptyList(), emptyMap());
29+
"theDataspaceContext", consumerProcessId, "theAgreementId", "theDatasetId",
30+
profile, emptyMap(), emptyList(), emptyMap());
3331
}
3432

35-
static @NonNull DataFlowStartMessage createStartMessage(String providerProcessId, URI callbackAddress, String transferType) {
36-
return createStartMessage(providerProcessId, callbackAddress, transferType, null);
33+
static @NonNull DataFlowStartMessage createStartMessage(String providerProcessId, String profile) {
34+
return createStartMessage(providerProcessId, profile, null);
3735
}
3836

39-
static @NonNull DataFlowStartMessage createStartMessage(String providerProcessId, URI callbackAddress, String transferType, DataAddress destinationDataAddress) {
37+
static @NonNull DataFlowStartMessage createStartMessage(String providerProcessId, String profile, DataAddress destinationDataAddress) {
4038
return new DataFlowStartMessage("theMessageId", "theParticipantId", "theCounterPartyId",
41-
"theDataspaceContext", providerProcessId, "theAgreementId", "theDatasetId", callbackAddress,
42-
transferType, destinationDataAddress, emptyMap(), emptyList(), emptyMap());
39+
"theDataspaceContext", providerProcessId, "theAgreementId", "theDatasetId",
40+
profile, destinationDataAddress, emptyMap(), emptyList(), emptyMap());
4341
}
4442
}

0 commit comments

Comments
 (0)