Skip to content

build: align with spec 1.0-RC3 #110

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
ndr-brt merged 1 commit into from
Jul 8, 2026
Merged

build: align with spec 1.0-RC3 #110

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
100 changes: 50 additions & 50 deletions dataplane-sdk-core/src/main/java/org/eclipse/dataplane/Dataplane.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,7 @@ public class Dataplane {
private ControlPlaneStore controlPlaneStore = new InMemoryControlPlaneStore(objectMapper);
private String id;
private URI endpoint;
private final Set<String> transferTypes = new HashSet<>();
private final Set<String> profiles = new HashSet<>();
private final Set<String> labels = new HashSet<>();

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

private Result<Void> checkControlPlane(String controlplaneId) {
if (controlPlaneStore.exists(controlplaneId)) {
return Result.success();
}
return Result.failure(new ControlPlaneNotRegistered(controlplaneId));
}

public Result<DataFlowStatusMessage> prepare(String controlplaneId, DataFlowPrepareMessage message) {
var initialDataFlow = DataFlow.newInstance()
.id(message.processId())
.state(DataFlow.State.INITIATING)
.labels(message.labels())
.metadata(message.metadata())
.callbackAddress(message.callbackAddress())
.transferType(message.transferType())
.datasetId(message.datasetId())
.agreementId(message.agreementId())
.participantId(message.participantId())
.counterPartyId(message.counterPartyId())
.dataspaceContext(message.dataspaceContext())
.controlplaneId(controlplaneId)
.type(DataFlow.Type.CONSUMER)
.build();

return checkControlPlane(controlplaneId)
.compose(v -> onPrepare.action(initialDataFlow))
return getControlPlane(controlplaneId)
.map(controlPlane -> DataFlow.newInstance()
.id(message.processId())
.state(DataFlow.State.INITIATING)
.labels(message.labels())
.metadata(message.metadata())
.callbackAddress(controlPlane.getEndpoint())
.profile(message.profile())
.datasetId(message.datasetId())
.agreementId(message.agreementId())
.participantId(message.participantId())
.counterPartyId(message.counterPartyId())
.dataspaceContext(message.dataspaceContext())
.controlplaneId(controlplaneId)
.type(DataFlow.Type.CONSUMER)
.build()
)
.compose(initialDataFlow -> onPrepare.action(initialDataFlow))
.compose(dataFlow -> {
if (dataFlow.isInitiating()) {
dataFlow.transitionToPrepared();
}

DataFlowStatusMessage response;
if (dataFlow.isPrepared() && dataFlow.isPush()) {
response = new DataFlowStatusMessage(dataFlow.getId(), initialDataFlow.getState().name(), dataFlow.getDataAddress(), null);
response = new DataFlowStatusMessage(dataFlow.getId(), dataFlow.getState().name(), dataFlow.getDataAddress(), null);
} else {
response = new DataFlowStatusMessage(dataFlow.getId(), initialDataFlow.getState().name(), null, null);
response = new DataFlowStatusMessage(dataFlow.getId(), dataFlow.getState().name(), null, null);
}

return save(dataFlow).map(it -> response);
});
}


public Result<DataFlowStatusMessage> start(String controlplaneId, DataFlowStartMessage message) {
var initialDataFlow = DataFlow.newInstance()
.id(message.processId())
.state(DataFlow.State.INITIATING)
.dataAddress(message.dataAddress())
.callbackAddress(message.callbackAddress())
.transferType(message.transferType())
.datasetId(message.datasetId())
.agreementId(message.agreementId())
.participantId(message.participantId())
.counterPartyId(message.counterPartyId())
.dataspaceContext(message.dataspaceContext())
.controlplaneId(controlplaneId)
.type(DataFlow.Type.PROVIDER)
.build();

return checkControlPlane(controlplaneId)
.compose(v -> onStart.action(initialDataFlow))
return getControlPlane(controlplaneId)
.map(controlPlane -> DataFlow.newInstance()
.id(message.processId())
.state(DataFlow.State.INITIATING)
.dataAddress(message.dataAddress())
.callbackAddress(controlPlane.getEndpoint())
.profile(message.profile())
.datasetId(message.datasetId())
.agreementId(message.agreementId())
.participantId(message.participantId())
.counterPartyId(message.counterPartyId())
.dataspaceContext(message.dataspaceContext())
.controlplaneId(controlplaneId)
.type(DataFlow.Type.PROVIDER)
.build()
)
.compose(initialDataFlow -> onStart.action(initialDataFlow))
.compose(dataFlow -> {
if (dataFlow.isInitiating()) {
dataFlow.transitionToStarted();
Expand Down Expand Up @@ -325,7 +317,7 @@ public Result<String> extractControlplaneId(String authorizationHeader) {

public Result<Void> registerOn(String controlPlaneEndpoint) {

var message = new DataPlaneRegistrationMessage(id, endpoint, transferTypes, labels);
var message = new DataPlaneRegistrationMessage(id, endpoint, profiles, labels);

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

private Result<ControlPlane> getControlPlane(String controlplaneId) {
var controlPlaneById = controlPlaneStore.findById(controlplaneId);
if (controlPlaneById.failed()) {
return Result.failure(new ControlPlaneNotRegistered(controlplaneId));
}
return controlPlaneById;
}

private DataAddress getDataAddressForResume(DataFlow dataFlow) {
if (dataFlow.isPull() && dataFlow.getType() == DataFlow.Type.PROVIDER) {
return dataFlow.getDataAddress();
Expand All @@ -363,7 +363,7 @@ private Result<Void> notifyControlPlane(String action, DataFlow dataFlow, Object
.header("content-type", "application/json")
.POST(HttpRequest.BodyPublishers.ofString(body));

controlPlaneStore.findById(dataFlow.getControlplaneId())
getControlPlane(dataFlow.getControlplaneId())
.compose(controlPlane -> {
var authorizationProfile = controlPlane.getAuthorization();
if (authorizationProfile != null) {
Expand Down Expand Up @@ -445,8 +445,8 @@ public Builder endpoint(URI endpoint) {
return this;
}

public Builder transferType(String transferType) {
dataplane.transferTypes.add(transferType);
public Builder profile(String profile) {
dataplane.profiles.add(profile);
return this;
}

Expand Down
18 changes: 9 additions & 9 deletions dataplane-sdk-core/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlow.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ public class DataFlow {

private String id;
private State state;
private String transferType;
private String profile;
private String datasetId;
private String agreementId;
private String participantId;
Expand Down Expand Up @@ -61,8 +61,8 @@ public URI getCallbackAddress() {
return callbackAddress;
}

public String getTransferType() {
return transferType;
public String getProfile() {
return profile;
}

public String getDatasetId() {
Expand Down Expand Up @@ -132,11 +132,11 @@ public void transitionToTerminated(String reason) {
}

public boolean isPush() {
return transferTypeLastToken().equalsIgnoreCase("push");
return profileLastToken().equalsIgnoreCase("push");
}

public boolean isPull() {
return transferTypeLastToken().equalsIgnoreCase("pull");
return profileLastToken().equalsIgnoreCase("pull");
}

public boolean isInitiating() {
Expand Down Expand Up @@ -167,8 +167,8 @@ public Type getType() {
return type;
}

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

public enum Type {
Expand Down Expand Up @@ -202,8 +202,8 @@ public Builder state(State state) {
return this;
}

public Builder transferType(String transferType) {
dataFlow.transferType = transferType;
public Builder profile(String profile) {
dataFlow.profile = profile;
return this;
}

Expand Down
4 changes: 1 addition & 3 deletions ...-sdk-core/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowPrepareMessage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,6 @@

package org.eclipse.dataplane.domain.dataflow;

import java.net.URI;
import java.util.List;
import java.util.Map;

Expand All @@ -26,8 +25,7 @@ public record DataFlowPrepareMessage(
String processId,
String agreementId,
String datasetId,
URI callbackAddress,
String transferType,
String profile,
Map<String, Object> claims,
List<String> labels,
Map<String, Object> metadata
Expand Down
4 changes: 1 addition & 3 deletions ...ne-sdk-core/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowStartMessage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@

import org.eclipse.dataplane.domain.DataAddress;

import java.net.URI;
import java.util.List;
import java.util.Map;

Expand All @@ -28,8 +27,7 @@ public record DataFlowStartMessage(
String processId,
String agreementId,
String datasetId,
URI callbackAddress,
String transferType,
String profile,
DataAddress dataAddress,
Map<String, Object> claims,
List<String> labels,
Expand Down
2 changes: 1 addition & 1 deletion ...src/main/java/org/eclipse/dataplane/domain/registration/DataPlaneRegistrationMessage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
public record DataPlaneRegistrationMessage(
String dataplaneId,
URI endpoint,
Set<String> transferTypes,
Set<String> profiles,
Set<String> labels
// TODO: authorization
) {
Expand Down
1 change: 1 addition & 0 deletions dataplane-sdk-core/src/main/java/org/eclipse/dataplane/port/store/ControlPlaneStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,5 +56,6 @@ public interface ControlPlaneStore {
* @param controlplaneId the id of the ControlPlane
* @return true, if the ControlPlane exists in the store, false otherwise
*/
@Deprecated(since = "1.0.0")
boolean exists(String controlplaneId);
}
2 changes: 1 addition & 1 deletion ...ane-sdk-core/src/testFixtures/java/org/eclipse/dataplane/store/DataFlowStoreTestBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,7 @@ private DataFlow dataFlow(String id) {
return DataFlow.newInstance()
.id(id)
.state(DataFlow.State.INITIATING)
.transferType("HTTP-PUSH")
.profile("HTTP-PUSH")
.datasetId("dataset")
.agreementId("agreement")
.participantId("participant")
Expand Down
4 changes: 2 additions & 2 deletions ...ostgresql/src/main/java/org/eclipse/dataplane/store/postgresql/PostgresDataFlowStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ public Result<Void> save(DataFlow dataFlow) {

try (var statement = connection.prepareStatement(upsertDataFlowTemplate())) {
statement.setString(1, dataFlow.getId());
statement.setString(2, dataFlow.getTransferType());
statement.setString(2, dataFlow.getProfile());
statement.setString(3, dataFlow.getType().name());
statement.setString(4, dataFlow.getState().name());
statement.setString(5, dataFlow.getDatasetId());
Expand Down Expand Up @@ -80,7 +80,7 @@ public Result<DataFlow> findById(String flowId) {
var dataFlow = DataFlow.newInstance()
.id(flowId)
.state(DataFlow.State.valueOf(resultSet.getString("state")))
.transferType(resultSet.getString("transfer_type"))
.profile(resultSet.getString("transfer_type"))
.datasetId(resultSet.getString("dataset_id"))
.agreementId(resultSet.getString("agreement_id"))
.participantId(resultSet.getString("participant_id"))
Expand Down
16 changes: 8 additions & 8 deletions e2e-tests/src/test/java/org/eclipse/dataplane/DataplaneTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,7 @@ void shouldFail_whenDataFlowDoesNotExist() {
@Test
void shouldReturnFailedFuture_whenControlPlaneIsNotAvailable() {
var dataplane = Dataplane.newInstance().onPrepare(Result::success).build();
dataplane.registerControlPlane(new ControlPlaneRegistrationMessage("controlplaneId", URI.create("http://localhost/any")));
dataplane.registerControlPlane(new ControlPlaneRegistrationMessage("controlplaneId", URI.create(controlPlane.baseUrl())));
dataplane.prepare("controlplaneId", createPrepareMessage());
controlPlane.stop();

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

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

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

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

var result = dataplane.notifyErrored("dataFlowId", new RuntimeException("some-error"));
Expand Down Expand Up @@ -161,7 +161,7 @@ void shouldRegisterOnTheControlPlane() {
var dataplane = Dataplane.newInstance()
.id("dataplane-id")
.endpoint(URI.create("http://localhost/dataplane"))
.transferType("SupportedTransferType-PUSH")
.profile("SupportedProfile-PUSH")
.label("label-one").label("label-two")
.build();

Expand All @@ -171,7 +171,7 @@ void shouldRegisterOnTheControlPlane() {
controlPlane.verify(putRequestedFor(urlPathEqualTo("/dataplanes"))
.withRequestBody(and(
matchingJsonPath("endpoint", equalTo("http://localhost/dataplane")),
matchingJsonPath("transferTypes[0]", equalTo("SupportedTransferType-PUSH")),
matchingJsonPath("profiles[0]", equalTo("SupportedProfile-PUSH")),
matchingJsonPath("labels.size()", equalTo("2"))
))
);
Expand All @@ -184,7 +184,7 @@ void shouldFail_whenStatusIsNot200() {
var dataplane = Dataplane.newInstance()
.id("dataplane-id")
.endpoint(URI.create("http://localhost/dataplane"))
.transferType("SupportedTransferType-PUSH")
.profile("SupportedProfile-PUSH")
.label("label-one").label("label-two")
.build();

Expand All @@ -196,6 +196,6 @@ void shouldFail_whenStatusIsNot200() {
}

private DataFlowPrepareMessage createPrepareMessage() {
return MessageFactory.createPrepareMessage("dataFlowId", URI.create(controlPlane.baseUrl()), "Something-PUSH");
return MessageFactory.createPrepareMessage("dataFlowId", "Something-PUSH");
}
}
18 changes: 8 additions & 10 deletions e2e-tests/src/test/java/org/eclipse/dataplane/MessageFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,26 +19,24 @@
import org.eclipse.dataplane.domain.dataflow.DataFlowStartMessage;
import org.jspecify.annotations.NonNull;

import java.net.URI;

import static java.util.Collections.emptyList;
import static java.util.Collections.emptyMap;

public interface MessageFactory {

static @NonNull DataFlowPrepareMessage createPrepareMessage(String consumerProcessId, URI callbackAddress, String transferType) {
static @NonNull DataFlowPrepareMessage createPrepareMessage(String consumerProcessId, String profile) {
return new DataFlowPrepareMessage("theMessageId", "theParticipantId", "theCounterPartyId",
"theDataspaceContext", consumerProcessId, "theAgreementId", "theDatasetId", callbackAddress,
transferType, emptyMap(), emptyList(), emptyMap());
"theDataspaceContext", consumerProcessId, "theAgreementId", "theDatasetId",
profile, emptyMap(), emptyList(), emptyMap());
}

static @NonNull DataFlowStartMessage createStartMessage(String providerProcessId, URI callbackAddress, String transferType) {
return createStartMessage(providerProcessId, callbackAddress, transferType, null);
static @NonNull DataFlowStartMessage createStartMessage(String providerProcessId, String profile) {
return createStartMessage(providerProcessId, profile, null);
}

static @NonNull DataFlowStartMessage createStartMessage(String providerProcessId, URI callbackAddress, String transferType, DataAddress destinationDataAddress) {
static @NonNull DataFlowStartMessage createStartMessage(String providerProcessId, String profile, DataAddress destinationDataAddress) {
return new DataFlowStartMessage("theMessageId", "theParticipantId", "theCounterPartyId",
"theDataspaceContext", providerProcessId, "theAgreementId", "theDatasetId", callbackAddress,
transferType, destinationDataAddress, emptyMap(), emptyList(), emptyMap());
"theDataspaceContext", providerProcessId, "theAgreementId", "theDatasetId",
profile, destinationDataAddress, emptyMap(), emptyList(), emptyMap());
}
}
Loading