diff --git a/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/Dataplane.java b/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/Dataplane.java index 0f0dfd1..7459083 100644 --- a/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/Dataplane.java +++ b/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/Dataplane.java @@ -75,7 +75,7 @@ public class Dataplane { private ControlPlaneStore controlPlaneStore = new InMemoryControlPlaneStore(objectMapper); private String id; private URI endpoint; - private final Set transferTypes = new HashSet<>(); + private final Set profiles = new HashSet<>(); private final Set labels = new HashSet<>(); private OnPrepare onPrepare = dataFlow -> Result.failure(new UnsupportedOperationException("onPrepare is not implemented")); @@ -106,32 +106,25 @@ public Result status(String dataFlowId) { .map(f -> new DataFlowStatusResponseMessage(f.getId(), f.getState().name())); } - private Result checkControlPlane(String controlplaneId) { - if (controlPlaneStore.exists(controlplaneId)) { - return Result.success(); - } - return Result.failure(new ControlPlaneNotRegistered(controlplaneId)); - } - public Result 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(); @@ -139,34 +132,33 @@ public Result prepare(String controlplaneId, DataFlowPrep 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 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(); @@ -325,7 +317,7 @@ public Result extractControlplaneId(String authorizationHeader) { public Result 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() @@ -344,6 +336,14 @@ public Result registerOn(String controlPlaneEndpoint) { }); } + private Result 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(); @@ -363,7 +363,7 @@ private Result 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) { @@ -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; } diff --git a/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlow.java b/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlow.java index 99055cf..cb36d1f 100644 --- a/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlow.java +++ b/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlow.java @@ -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; @@ -61,8 +61,8 @@ public URI getCallbackAddress() { return callbackAddress; } - public String getTransferType() { - return transferType; + public String getProfile() { + return profile; } public String getDatasetId() { @@ -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() { @@ -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 { @@ -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; } diff --git a/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowPrepareMessage.java b/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowPrepareMessage.java index 6fb9228..9b05e46 100644 --- a/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowPrepareMessage.java +++ b/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowPrepareMessage.java @@ -14,7 +14,6 @@ package org.eclipse.dataplane.domain.dataflow; -import java.net.URI; import java.util.List; import java.util.Map; @@ -26,8 +25,7 @@ public record DataFlowPrepareMessage( String processId, String agreementId, String datasetId, - URI callbackAddress, - String transferType, + String profile, Map claims, List labels, Map metadata diff --git a/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowStartMessage.java b/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowStartMessage.java index c9ff11e..cd2e604 100644 --- a/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowStartMessage.java +++ b/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowStartMessage.java @@ -16,7 +16,6 @@ import org.eclipse.dataplane.domain.DataAddress; -import java.net.URI; import java.util.List; import java.util.Map; @@ -28,8 +27,7 @@ public record DataFlowStartMessage( String processId, String agreementId, String datasetId, - URI callbackAddress, - String transferType, + String profile, DataAddress dataAddress, Map claims, List labels, diff --git a/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/domain/registration/DataPlaneRegistrationMessage.java b/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/domain/registration/DataPlaneRegistrationMessage.java index d165472..f5d8a20 100644 --- a/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/domain/registration/DataPlaneRegistrationMessage.java +++ b/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/domain/registration/DataPlaneRegistrationMessage.java @@ -20,7 +20,7 @@ public record DataPlaneRegistrationMessage( String dataplaneId, URI endpoint, - Set transferTypes, + Set profiles, Set labels // TODO: authorization ) { diff --git a/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/port/store/ControlPlaneStore.java b/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/port/store/ControlPlaneStore.java index dbd3131..c3ae468 100644 --- a/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/port/store/ControlPlaneStore.java +++ b/dataplane-sdk-core/src/main/java/org/eclipse/dataplane/port/store/ControlPlaneStore.java @@ -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); } diff --git a/dataplane-sdk-core/src/testFixtures/java/org/eclipse/dataplane/store/DataFlowStoreTestBase.java b/dataplane-sdk-core/src/testFixtures/java/org/eclipse/dataplane/store/DataFlowStoreTestBase.java index 99650ec..7e9a5eb 100644 --- a/dataplane-sdk-core/src/testFixtures/java/org/eclipse/dataplane/store/DataFlowStoreTestBase.java +++ b/dataplane-sdk-core/src/testFixtures/java/org/eclipse/dataplane/store/DataFlowStoreTestBase.java @@ -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") diff --git a/dataplane-sdk-postgresql/src/main/java/org/eclipse/dataplane/store/postgresql/PostgresDataFlowStore.java b/dataplane-sdk-postgresql/src/main/java/org/eclipse/dataplane/store/postgresql/PostgresDataFlowStore.java index 45ed157..30ba8a2 100644 --- a/dataplane-sdk-postgresql/src/main/java/org/eclipse/dataplane/store/postgresql/PostgresDataFlowStore.java +++ b/dataplane-sdk-postgresql/src/main/java/org/eclipse/dataplane/store/postgresql/PostgresDataFlowStore.java @@ -40,7 +40,7 @@ public Result 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()); @@ -80,7 +80,7 @@ public Result 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")) diff --git a/e2e-tests/src/test/java/org/eclipse/dataplane/DataplaneTest.java b/e2e-tests/src/test/java/org/eclipse/dataplane/DataplaneTest.java index d43acfa..c28fd67 100644 --- a/e2e-tests/src/test/java/org/eclipse/dataplane/DataplaneTest.java +++ b/e2e-tests/src/test/java/org/eclipse/dataplane/DataplaneTest.java @@ -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(); @@ -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"); @@ -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"); @@ -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")); @@ -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(); @@ -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")) )) ); @@ -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(); @@ -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"); } } diff --git a/e2e-tests/src/test/java/org/eclipse/dataplane/MessageFactory.java b/e2e-tests/src/test/java/org/eclipse/dataplane/MessageFactory.java index ba21d31..0248872 100644 --- a/e2e-tests/src/test/java/org/eclipse/dataplane/MessageFactory.java +++ b/e2e-tests/src/test/java/org/eclipse/dataplane/MessageFactory.java @@ -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()); } } diff --git a/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/AuthorizationOauth2Test.java b/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/AuthorizationOauth2Test.java index 2c3857a..722a59b 100644 --- a/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/AuthorizationOauth2Test.java +++ b/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/AuthorizationOauth2Test.java @@ -103,10 +103,10 @@ void shouldCommunicateWithControlPlaneUsingOauth2Authorization() { ); dataPlane.registerControlPlane(controlPlaneRegistrationMessage).orElseThrow(RuntimeException::new); - var transferType = "FileSystemAsync-PUSH"; + var profile = "FileSystemAsync-PUSH"; var processId = UUID.randomUUID().toString(); var consumerProcessId = "consumer_" + processId; - var prepareMessage = createPrepareMessage(consumerProcessId, controlPlane.consumerCallbackAddress(), transferType); + var prepareMessage = createPrepareMessage(consumerProcessId, profile); controlPlane.consumerPrepare(prepareMessage).statusCode(202).extract().as(DataFlowStatusMessage.class); diff --git a/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/AuthorizationTest.java b/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/AuthorizationTest.java index 8a1e60b..0772cb6 100644 --- a/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/AuthorizationTest.java +++ b/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/AuthorizationTest.java @@ -75,7 +75,7 @@ void shouldCommunicateWithControlPlaneUsingRegisteredAuthorization() { dataPlane.registerControlPlane(controlPlaneRegistrationMessage).orElseThrow(RuntimeException::new); var consumerProcessId = "consumer_" + UUID.randomUUID(); - var prepareMessage = createPrepareMessage(consumerProcessId, controlPlane.consumerCallbackAddress(), "FileSystemAsync-PUSH"); + var prepareMessage = createPrepareMessage(consumerProcessId, "FileSystemAsync-PUSH"); controlPlane.consumerPrepare(prepareMessage).statusCode(202).extract().as(DataFlowStatusMessage.class); @@ -97,7 +97,7 @@ void shouldGetUnauthorized_whenControlPlaneIsNotAuthenticated() { dataPlane.registerControlPlane(controlPlaneRegistrationMessage).orElseThrow(RuntimeException::new); var consumerProcessId = "consumer_" + UUID.randomUUID(); - var prepareMessage = createPrepareMessage(consumerProcessId, controlPlane.consumerCallbackAddress(), "FileSystemAsync-PUSH"); + var prepareMessage = createPrepareMessage(consumerProcessId, "FileSystemAsync-PUSH"); controlPlane.consumerPrepare(prepareMessage).statusCode(401); } @@ -114,7 +114,7 @@ void shouldGetUnauthorized_withDataPlaneIsNotAuthenticated() { dataPlane.registerControlPlane(controlPlaneRegistrationMessage).orElseThrow(RuntimeException::new); var consumerProcessId = "consumer_" + UUID.randomUUID(); - var prepareMessage = createPrepareMessage(consumerProcessId, controlPlane.consumerCallbackAddress(), "FileSystemAsync-PUSH"); + var prepareMessage = createPrepareMessage(consumerProcessId, "FileSystemAsync-PUSH"); controlPlane.consumerPrepare(prepareMessage).statusCode(202).extract().as(DataFlowStatusMessage.class); diff --git a/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/ConsumerPullTest.java b/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/ConsumerPullTest.java index 8127429..c0d439f 100644 --- a/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/ConsumerPullTest.java +++ b/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/ConsumerPullTest.java @@ -32,7 +32,6 @@ import java.io.File; import java.io.IOException; -import java.net.URI; import java.nio.file.Files; import java.nio.file.Path; import java.util.Objects; @@ -57,13 +56,15 @@ class ConsumerPullTest { private final ControlPlane controlPlane = ControlPlane.newInstance() .authorizationTokenGenerator(() -> TOKEN_GENERATOR.apply("control-plane-id")) .build(); - private final ConsumerDataPlane consumerDataPlane = new ConsumerDataPlane(); - private final ProviderDataPlane providerDataPlane = new ProviderDataPlane(filesAvailableOnProvider); + private ConsumerDataPlane consumerDataPlane; + private ProviderDataPlane providerDataPlane; @BeforeEach void setUp() { httpServer.start(); controlPlane.initialize(httpServer, "/consumer/data-plane", "/provider/data-plane"); + consumerDataPlane = new ConsumerDataPlane(); + providerDataPlane = new ProviderDataPlane(filesAvailableOnProvider); httpServer.deploy("/consumer/data-plane", consumerDataPlane.controller()); httpServer.deploy("/provider/data-plane", providerDataPlane.controller()); @@ -76,17 +77,17 @@ void tearDown() { @Test void shouldPullDataFromProvider() { - var transferType = "FileSystem-PULL"; + var profile = "FileSystem-PULL"; var processId = UUID.randomUUID().toString(); var consumerProcessId = "consumer_" + processId; - var prepareMessage = createPrepareMessage(consumerProcessId, controlPlane.consumerCallbackAddress(), transferType); + var prepareMessage = createPrepareMessage(consumerProcessId, profile); var prepareResponse = controlPlane.consumerPrepare(prepareMessage).statusCode(200).extract().as(DataFlowStatusMessage.class); assertThat(prepareResponse.state()).isEqualTo(PREPARED.name()); assertThat(prepareResponse.dataAddress()).isNull(); var providerProcessId = "provider_" + processId; - var startMessage = createStartMessage(providerProcessId, controlPlane.providerCallbackAddress(), transferType); + var startMessage = createStartMessage(providerProcessId, profile); var startResponse = controlPlane.providerStart(startMessage).statusCode(200).extract().as(DataFlowStatusMessage.class); assertThat(startResponse.state()).isEqualTo(STARTED.name()); assertThat(startResponse.dataAddress()).isNotNull(); @@ -101,14 +102,14 @@ void shouldPullDataFromProvider() { @Test void shouldPermitAsyncStartup() { - var transferType = "FileSystemAsync-PULL"; + var profile = "FileSystemAsync-PULL"; var processId = UUID.randomUUID().toString(); var consumerProcessId = "consumer_" + processId; - var prepareMessage = createPrepareMessage(consumerProcessId, controlPlane.consumerCallbackAddress(), transferType); + var prepareMessage = createPrepareMessage(consumerProcessId, profile); controlPlane.consumerPrepare(prepareMessage).statusCode(200).extract().as(DataFlowStatusMessage.class); var providerProcessId = "provider_" + processId; - var startMessage = createStartMessage(providerProcessId, controlPlane.providerCallbackAddress(), transferType); + var startMessage = createStartMessage(providerProcessId, profile); var startResponse = controlPlane.providerStart(startMessage).statusCode(202).extract().as(DataFlowStatusMessage.class); assertThat(startResponse.state()).isEqualTo(STARTING.name()); assertThat(startResponse.dataAddress()).isNull(); @@ -131,7 +132,7 @@ private class ConsumerDataPlane { ConsumerDataPlane() { - sdk.registerControlPlane(new ControlPlaneRegistrationMessage("control-plane-id", URI.create("http://localhost:any"), createAuthorizationProfile("consumer"))); + sdk.registerControlPlane(new ControlPlaneRegistrationMessage("control-plane-id", controlPlane.consumerCallbackAddress(), createAuthorizationProfile("consumer"))); try { storage = Files.createTempDirectory("consumer-storage"); } catch (IOException e) { @@ -167,7 +168,7 @@ private class ProviderDataPlane { private final int filesToBeCreated; ProviderDataPlane(int fileToBeCreated) { - sdk.registerControlPlane(new ControlPlaneRegistrationMessage("control-plane-id", URI.create("http://localhost:any"), createAuthorizationProfile("provider"))); + sdk.registerControlPlane(new ControlPlaneRegistrationMessage("control-plane-id", controlPlane.providerCallbackAddress(), createAuthorizationProfile("provider"))); this.filesToBeCreated = fileToBeCreated; } @@ -182,7 +183,7 @@ public void completeStartup(String dataFlowId) { } private Result onStart(DataFlow dataFlow) { - if (dataFlow.getTransferType().equals("FileSystemAsync-PULL")) { + if (dataFlow.getProfile().equals("FileSystemAsync-PULL")) { dataFlow.transitionToStarting(); return Result.success(dataFlow); } diff --git a/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/ProviderPushTest.java b/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/ProviderPushTest.java index 744b6f4..41971fb 100644 --- a/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/ProviderPushTest.java +++ b/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/ProviderPushTest.java @@ -31,7 +31,6 @@ import org.junit.jupiter.api.Test; import java.io.IOException; -import java.net.URI; import java.nio.file.Files; import java.nio.file.Path; import java.util.UUID; @@ -60,13 +59,15 @@ public class ProviderPushTest { .authorizationTokenGenerator(() -> TOKEN_GENERATOR.apply("control-plane-id")) .build(); - private final ConsumerDataPlane consumerDataPlane = new ConsumerDataPlane(); - private final ProviderDataPlane providerDataPlane = new ProviderDataPlane(); + private ConsumerDataPlane consumerDataPlane; + private ProviderDataPlane providerDataPlane; @BeforeEach void setUp() { httpServer.start(); controlPlane.initialize(httpServer, "/consumer/data-plane", "/provider/data-plane"); + consumerDataPlane = new ConsumerDataPlane(controlPlane); + providerDataPlane = new ProviderDataPlane(controlPlane); httpServer.deploy("/consumer/data-plane", consumerDataPlane.controller()); httpServer.deploy("/provider/data-plane", providerDataPlane.controller()); @@ -79,10 +80,10 @@ void tearDown() { @Test void shouldPushDataToEndpointPreparedByConsumer() { - var transferType = "FileSystem-PUSH"; + var profile = "FileSystem-PUSH"; var processId = UUID.randomUUID().toString(); var consumerProcessId = "consumer_" + processId; - var prepareMessage = createPrepareMessage(consumerProcessId, controlPlane.consumerCallbackAddress(), transferType); + var prepareMessage = createPrepareMessage(consumerProcessId, profile); var prepareResponse = controlPlane.consumerPrepare(prepareMessage).statusCode(200).extract().as(DataFlowStatusMessage.class); assertThat(prepareResponse.state()).isEqualTo(PREPARED.name()); @@ -90,7 +91,7 @@ void shouldPushDataToEndpointPreparedByConsumer() { var destinationDataAddress = prepareResponse.dataAddress(); var providerProcessId = "provider_" + processId; - var startMessage = createStartMessage(providerProcessId, controlPlane.providerCallbackAddress(), transferType, destinationDataAddress); + var startMessage = createStartMessage(providerProcessId, profile, destinationDataAddress); var startResponse = controlPlane.providerStart(startMessage).statusCode(200).extract().as(DataFlowStatusMessage.class); assertThat(startResponse.state()).isEqualTo(STARTED.name()); @@ -110,16 +111,16 @@ void shouldPushDataToEndpointPreparedByConsumer() { @Test void shouldSendError_whenFlowFails() { - var transferType = "FileSystem-PUSH"; + var profile = "FileSystem-PUSH"; var processId = UUID.randomUUID().toString(); var consumerProcessId = "consumer_" + processId; - var prepareMessage = createPrepareMessage(consumerProcessId, controlPlane.consumerCallbackAddress(), transferType); + var prepareMessage = createPrepareMessage(consumerProcessId, profile); controlPlane.consumerPrepare(prepareMessage).statusCode(200).extract().as(DataFlowStatusMessage.class); var invalidDataAddress = new DataAddress("FileSystem", "", emptyList()); var providerProcessId = "provider_" + processId; - var startMessage = createStartMessage(providerProcessId, controlPlane.providerCallbackAddress(), transferType, invalidDataAddress); + var startMessage = createStartMessage(providerProcessId, profile, invalidDataAddress); controlPlane.providerStart(startMessage).statusCode(200).extract().as(DataFlowStatusMessage.class); await().untilAsserted(() -> { @@ -133,10 +134,10 @@ void shouldSendError_whenFlowFails() { @Test void shouldPermitAsyncPreparation() { - var transferType = "FileSystemAsync-PUSH"; + var profile = "FileSystemAsync-PUSH"; var processId = UUID.randomUUID().toString(); var consumerProcessId = "consumer_" + processId; - var prepareMessage = createPrepareMessage(consumerProcessId, controlPlane.consumerCallbackAddress(), transferType); + var prepareMessage = createPrepareMessage(consumerProcessId, profile); var prepareResponse = controlPlane.consumerPrepare(prepareMessage).statusCode(202).extract().as(DataFlowStatusMessage.class); assertThat(prepareResponse.state()).isEqualTo(PREPARING.name()); @@ -157,8 +158,9 @@ private static class ProviderDataPlane { .onStart(this::onStart) .build(); - ProviderDataPlane() { - sdk.registerControlPlane(new ControlPlaneRegistrationMessage("control-plane-id", URI.create("http://localhost:any"), createAuthorizationProfile("provider"))); + ProviderDataPlane(ControlPlane controlPlane) { + sdk.registerControlPlane(new ControlPlaneRegistrationMessage("control-plane-id", + controlPlane.providerCallbackAddress(), createAuthorizationProfile("provider"))); } private Result onStart(DataFlow dataFlow) { @@ -199,8 +201,9 @@ private static class ConsumerDataPlane { .onTerminate(Result::success) .build(); - ConsumerDataPlane() { - sdk.registerControlPlane(new ControlPlaneRegistrationMessage("control-plane-id", URI.create("http://localhost:any"), createAuthorizationProfile("consumer"))); + ConsumerDataPlane(ControlPlane controlPlane) { + sdk.registerControlPlane(new ControlPlaneRegistrationMessage("control-plane-id", + controlPlane.consumerCallbackAddress(), createAuthorizationProfile("consumer"))); } public void completePreparation(String dataFlowId) { @@ -210,7 +213,7 @@ public void completePreparation(String dataFlowId) { } private Result onPrepare(DataFlow dataFlow) { - if (dataFlow.getTransferType().equals("FileSystemAsync-PUSH")) { + if (dataFlow.getProfile().equals("FileSystemAsync-PUSH")) { dataFlow.transitionToPreparing(); return Result.success(dataFlow); } diff --git a/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/StreamingPullTest.java b/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/StreamingPullTest.java index 25acf10..090bc38 100644 --- a/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/StreamingPullTest.java +++ b/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/StreamingPullTest.java @@ -83,17 +83,17 @@ void tearDown() { @Test void shouldPullDataFromProvider_thenProviderTerminatesIt() { - var transferType = "FileSystemStreaming-PULL"; + var profile = "FileSystemStreaming-PULL"; var processId = UUID.randomUUID().toString(); var consumerProcessId = "consumer_" + processId; - var prepareMessage = createPrepareMessage(consumerProcessId, controlPlane.consumerCallbackAddress(), transferType); + var prepareMessage = createPrepareMessage(consumerProcessId, profile); var prepareResponse = controlPlane.consumerPrepare(prepareMessage).statusCode(200).extract().as(DataFlowStatusMessage.class); assertThat(prepareResponse.state()).isEqualTo(PREPARED.name()); assertThat(prepareResponse.dataAddress()).isNull(); var providerProcessId = "provider_" + processId; - var startMessage = createStartMessage(providerProcessId, controlPlane.providerCallbackAddress(), transferType); + var startMessage = createStartMessage(providerProcessId, profile); var startResponse = controlPlane.providerStart(startMessage).statusCode(200).extract().as(DataFlowStatusMessage.class); assertThat(startResponse.state()).isEqualTo(STARTED.name()); assertThat(startResponse.dataAddress()).isNotNull(); @@ -113,17 +113,17 @@ void shouldPullDataFromProvider_thenProviderTerminatesIt() { @Test void shouldSuspendAndResumeOnProvider() { - var transferType = "FileSystemStreaming-PULL"; + var profile = "FileSystemStreaming-PULL"; var processId = UUID.randomUUID().toString(); var consumerProcessId = "consumer_" + processId; - var prepareMessage = createPrepareMessage(consumerProcessId, controlPlane.consumerCallbackAddress(), transferType); + var prepareMessage = createPrepareMessage(consumerProcessId, profile); var prepareResponse = controlPlane.consumerPrepare(prepareMessage).statusCode(200).extract().as(DataFlowStatusMessage.class); assertThat(prepareResponse.state()).isEqualTo(PREPARED.name()); assertThat(prepareResponse.dataAddress()).isNull(); var providerProcessId = "provider_" + processId; - var startMessage = createStartMessage(providerProcessId, controlPlane.providerCallbackAddress(), transferType); + var startMessage = createStartMessage(providerProcessId, profile); var startResponse = controlPlane.providerStart(startMessage).statusCode(200).extract().as(DataFlowStatusMessage.class); assertThat(startResponse.state()).isEqualTo(STARTED.name()); assertThat(startResponse.dataAddress()).isNotNull(); diff --git a/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/StreamingPushTest.java b/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/StreamingPushTest.java index da101de..537a025 100644 --- a/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/StreamingPushTest.java +++ b/e2e-tests/src/test/java/org/eclipse/dataplane/scenario/StreamingPushTest.java @@ -78,10 +78,10 @@ void tearDown() { @Test void shouldPushDataToEndpointPreparedByConsumer() { - var transferType = "FileSystemStreaming-PUSH"; + var profile = "FileSystemStreaming-PUSH"; var processId = UUID.randomUUID().toString(); var consumerProcessId = "consumer_" + processId; - var prepareMessage = MessageFactory.createPrepareMessage(consumerProcessId, URI.create("http://callback"), transferType); + var prepareMessage = MessageFactory.createPrepareMessage(consumerProcessId, profile); var prepareResponse = controlPlane.consumerPrepare(prepareMessage).statusCode(200).extract().as(DataFlowStatusMessage.class); assertThat(prepareResponse.state()).isEqualTo(PREPARED.name()); @@ -89,7 +89,7 @@ void shouldPushDataToEndpointPreparedByConsumer() { var destinationDataAddress = prepareResponse.dataAddress(); var providerProcessId = "provider_" + processId; - var startMessage = MessageFactory.createStartMessage(providerProcessId, controlPlane.providerCallbackAddress(), transferType, destinationDataAddress); + var startMessage = MessageFactory.createStartMessage(providerProcessId, profile, destinationDataAddress); var startResponse = controlPlane.providerStart(startMessage).statusCode(200).extract().as(DataFlowStatusMessage.class); assertThat(startResponse.state()).isEqualTo(STARTED.name()); @@ -100,10 +100,10 @@ void shouldPushDataToEndpointPreparedByConsumer() { @Test void shouldSuspendAndResumeByConsumer() { - var transferType = "FileSystemStreaming-PUSH"; + var profile = "FileSystemStreaming-PUSH"; var processId = UUID.randomUUID().toString(); var consumerProcessId = "consumer_" + processId; - var prepareMessage = MessageFactory.createPrepareMessage(consumerProcessId, URI.create("http://callback"), transferType); + var prepareMessage = MessageFactory.createPrepareMessage(consumerProcessId, profile); var prepareResponse = controlPlane.consumerPrepare(prepareMessage).statusCode(200).extract().as(DataFlowStatusMessage.class); assertThat(prepareResponse.state()).isEqualTo(PREPARED.name()); @@ -111,7 +111,7 @@ void shouldSuspendAndResumeByConsumer() { var destinationDataAddress = prepareResponse.dataAddress(); var providerProcessId = "provider_" + processId; - var startMessage = MessageFactory.createStartMessage(providerProcessId, controlPlane.providerCallbackAddress(), transferType, destinationDataAddress); + var startMessage = MessageFactory.createStartMessage(providerProcessId, profile, destinationDataAddress); var startResponse = controlPlane.providerStart(startMessage).statusCode(200).extract().as(DataFlowStatusMessage.class); assertThat(startResponse.state()).isEqualTo(STARTED.name()); diff --git a/e2e-tests/src/test/java/org/eclipse/dataplane/tck/DpsTckTest.java b/e2e-tests/src/test/java/org/eclipse/dataplane/tck/DpsTckTest.java index 58a6498..808a728 100644 --- a/e2e-tests/src/test/java/org/eclipse/dataplane/tck/DpsTckTest.java +++ b/e2e-tests/src/test/java/org/eclipse/dataplane/tck/DpsTckTest.java @@ -30,6 +30,7 @@ import java.util.stream.Collectors; import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.platform.launcher.TagFilter.excludeTags; /** * Runs the DPS TCK data plane verification tests against the dataplane-sdk-java HTTP server. @@ -50,7 +51,7 @@ public class DpsTckTest { void startDataplane() { var tckControlPlaneId = "tck-control-plane"; var tckDataplane = new TckDataplane(tckControlPlaneId); - tckDataplane.getDataplane().registerControlPlane(new ControlPlaneRegistrationMessage(tckControlPlaneId, URI.create("http://localhost"))); + tckDataplane.getDataplane().registerControlPlane(new ControlPlaneRegistrationMessage(tckControlPlaneId, URI.create("http://localhost:8083"))); httpServer = new HttpServer(); httpServer.start(); @@ -74,6 +75,7 @@ void runDpsTckTests() { .launcher(DpsSystemLauncher.class) .properties(properties) .addPackage("org.eclipse.dataspacetck.dps.verification.dataplane") + .filters(excludeTags("http-profile")) .build() .execute(); diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 8f7820f..165152b 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -14,7 +14,7 @@ nimbusJoseJwt = "10.9.1" postgresql = "42.7.12" restAssured = "6.0.0" slf4j = "2.0.18" -tck-dps = "1.1.2" +tck-dps = "1.2.0" testcontainers = "2.0.5" wiremock = "3.13.2" diff --git a/settings.gradle.kts b/settings.gradle.kts index 6f876b4..27f62e9 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -3,6 +3,7 @@ rootProject.name = "dataplane-sdk" dependencyResolutionManagement { repositories { mavenCentral() + mavenLocal() } }