refactor: make callback address URI (#55)

ronjaquensel · web-flow · commit 5051fe64991a · 2026-03-24T16:30:03.000+01:00
diff --git a/src/main/java/org/eclipse/dataplane/Dataplane.java b/src/main/java/org/eclipse/dataplane/Dataplane.java
@@ -65,7 +65,7 @@ public class Dataplane {
     private final DataFlowStore dataFlowStore = new InMemoryDataFlowStore(objectMapper);
     private final ControlPlaneStore controlPlaneStore = new InMemoryControlPlaneStore(objectMapper);
     private String id;
-    private String endpoint;
+    private URI endpoint;
     private final Set<String> transferTypes = new HashSet<>();
     private final Set<String> labels = new HashSet<>();
 
@@ -306,7 +306,7 @@ private Result<Void> notifyControlPlane(String action, DataFlow dataFlow, Object
                 .map(body -> {
                     var endpoint = dataFlow.callbackEndpointFor(action);
                     var requestBuilder = HttpRequest.newBuilder()
-                            .uri(URI.create(endpoint))
+                            .uri(endpoint)
                             .header("content-type", "application/json")
                             .POST(HttpRequest.BodyPublishers.ofString(body));
 
@@ -384,7 +384,7 @@ public Builder id(String id) {
             return this;
         }
 
-        public Builder endpoint(String endpoint) {
+        public Builder endpoint(URI endpoint) {
             dataplane.endpoint = endpoint;
             return this;
         }
diff --git a/src/main/java/org/eclipse/dataplane/domain/controlplane/ControlPlane.java b/src/main/java/org/eclipse/dataplane/domain/controlplane/ControlPlane.java
@@ -17,21 +17,22 @@
 import com.fasterxml.jackson.databind.annotation.JsonPOJOBuilder;
 import org.eclipse.dataplane.domain.registration.AuthorizationProfile;
 
+import java.net.URI;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
 
 public class ControlPlane {
 
     private String id;
-    private String endpoint;
+    private URI endpoint;
     private final List<AuthorizationProfile> authorizations = new ArrayList<>();
 
     public String getId() {
         return id;
     }
 
-    public String getEndpoint() {
+    public URI getEndpoint() {
         return endpoint;
     }
 
@@ -66,7 +67,7 @@ public Builder id(String id) {
             return this;
         }
 
-        public Builder endpoint(String endpoint) {
+        public Builder endpoint(URI endpoint) {
             controlPlane.endpoint = endpoint;
             return this;
         }
diff --git a/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlow.java b/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlow.java
@@ -17,6 +17,7 @@
 
 import org.eclipse.dataplane.domain.DataAddress;
 
+import java.net.URI;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -32,7 +33,7 @@ public class DataFlow {
     private String participantId;
     private String counterPartyId;
     private String dataspaceContext;
-    private String callbackAddress;
+    private URI callbackAddress;
     private String suspensionReason;
     private String terminationReason;
     private List<String> labels;
@@ -55,7 +56,7 @@ public DataAddress getDataAddress() {
         return dataAddress;
     }
 
-    public String getCallbackAddress() {
+    public URI getCallbackAddress() {
         return callbackAddress;
     }
 
@@ -153,8 +154,8 @@ public void setDataAddress(DataAddress dataAddress) {
         this.dataAddress = dataAddress;
     }
 
-    public String callbackEndpointFor(String action) {
-        return getCallbackAddress() + "/transfers/" + getId() + "/dataflow/" + action;
+    public URI callbackEndpointFor(String action) {
+        return URI.create(getCallbackAddress() + "/transfers/" + getId() + "/dataflow/" + action);
     }
 
     public static class Builder {
@@ -224,7 +225,7 @@ public Builder dataAddress(DataAddress dataAddress) {
             return this;
         }
 
-        public Builder callbackAddress(String callbackAddress) {
+        public Builder callbackAddress(URI callbackAddress) {
             dataFlow.callbackAddress = callbackAddress;
             return this;
         }
diff --git a/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowPrepareMessage.java b/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowPrepareMessage.java
@@ -14,6 +14,7 @@
 
 package org.eclipse.dataplane.domain.dataflow;
 
+import java.net.URI;
 import java.util.List;
 import java.util.Map;
 
@@ -25,7 +26,7 @@ public record DataFlowPrepareMessage(
         String processId,
         String agreementId,
         String datasetId,
-        String callbackAddress, // TODO: make URI!
+        URI callbackAddress,
         String transferType,
         List<String> labels,
         Map<String, Object> metadata
diff --git a/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowStartMessage.java b/src/main/java/org/eclipse/dataplane/domain/dataflow/DataFlowStartMessage.java
@@ -16,6 +16,7 @@
 
 import org.eclipse.dataplane.domain.DataAddress;
 
+import java.net.URI;
 import java.util.List;
 import java.util.Map;
 
@@ -27,7 +28,7 @@ public record DataFlowStartMessage(
         String processId,
         String agreementId,
         String datasetId,
-        String callbackAddress,
+        URI callbackAddress,
         String transferType,
         DataAddress dataAddress,
         List<String> labels,
diff --git a/src/main/java/org/eclipse/dataplane/domain/registration/ControlPlaneRegistrationMessage.java b/src/main/java/org/eclipse/dataplane/domain/registration/ControlPlaneRegistrationMessage.java
@@ -14,16 +14,17 @@
 
 package org.eclipse.dataplane.domain.registration;
 
+import java.net.URI;
 import java.util.List;
 
 import static java.util.Collections.emptyList;
 
 public record ControlPlaneRegistrationMessage(
         String controlplaneId,
-        String endpoint,
+        URI endpoint,
         List<AuthorizationProfile> authorization
 ) {
-    public ControlPlaneRegistrationMessage(String controlplaneId, String endpoint) {
+    public ControlPlaneRegistrationMessage(String controlplaneId, URI endpoint) {
         this(controlplaneId, endpoint, emptyList());
     }
 }
diff --git a/src/main/java/org/eclipse/dataplane/domain/registration/DataPlaneRegistrationMessage.java b/src/main/java/org/eclipse/dataplane/domain/registration/DataPlaneRegistrationMessage.java
@@ -14,11 +14,12 @@
 
 package org.eclipse.dataplane.domain.registration;
 
+import java.net.URI;
 import java.util.Set;
 
 public record DataPlaneRegistrationMessage(
         String dataplaneId,
-        String endpoint,
+        URI endpoint,
         Set<String> transferTypes,
         Set<String> labels
 // TODO: authorization
diff --git a/src/main/java/org/eclipse/dataplane/port/store/ControlPlaneStore.java b/src/main/java/org/eclipse/dataplane/port/store/ControlPlaneStore.java
@@ -17,12 +17,14 @@
 import org.eclipse.dataplane.domain.Result;
 import org.eclipse.dataplane.domain.controlplane.ControlPlane;
 
+import java.net.URI;
+
 public interface ControlPlaneStore {
     Result<Void> save(ControlPlane controlPlane);
 
     Result<ControlPlane> findById(String controlplaneId);
 
     Result<Void> delete(String id);
 
-    Result<ControlPlane> findByEndpoint(String endpoint);
+    Result<ControlPlane> findByEndpoint(URI endpoint);
 }
diff --git a/src/main/java/org/eclipse/dataplane/port/store/InMemoryControlPlaneStore.java b/src/main/java/org/eclipse/dataplane/port/store/InMemoryControlPlaneStore.java
@@ -20,6 +20,7 @@
 import org.eclipse.dataplane.domain.controlplane.ControlPlane;
 import org.eclipse.dataplane.port.exception.ResourceNotFoundException;
 
+import java.net.URI;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
@@ -63,7 +64,7 @@ public Result<Void> delete(String id) {
     }
 
     @Override
-    public Result<ControlPlane> findByEndpoint(String endpoint) {
+    public Result<ControlPlane> findByEndpoint(URI endpoint) {
         return store.values().stream().map(this::deserialize).filter(Result::succeeded)
                 .map(Result::getContent).filter(it -> Objects.equals(endpoint, it.getEndpoint()))
                 .findAny().map(Result::success)
diff --git a/src/test/java/org/eclipse/dataplane/ControlPlane.java b/src/test/java/org/eclipse/dataplane/ControlPlane.java
@@ -29,6 +29,7 @@
 import org.eclipse.dataplane.domain.dataflow.DataFlowSuspendMessage;
 import org.eclipse.dataplane.domain.dataflow.DataFlowTerminateMessage;
 
+import java.net.URI;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.function.Predicate;
@@ -87,12 +88,12 @@ public ValidatableResponse providerTerminate(String dataFlowId, DataFlowTerminat
         return providerClient.terminate(dataFlowId, terminateMessage);
     }
 
-    public String providerCallbackAddress() {
-        return "http://localhost:%d/provider/control-plane".formatted(httpServer.port());
+    public URI providerCallbackAddress() {
+        return URI.create("http://localhost:%d/provider/control-plane".formatted(httpServer.port()));
     }
 
-    public String consumerCallbackAddress() {
-        return "http://localhost:%d/consumer/control-plane".formatted(httpServer.port());
+    public URI consumerCallbackAddress() {
+        return URI.create("http://localhost:%d/consumer/control-plane".formatted(httpServer.port()));
     }
 
     @Deprecated(forRemoval = true)
diff --git a/src/test/java/org/eclipse/dataplane/DataplaneTest.java b/src/test/java/org/eclipse/dataplane/DataplaneTest.java
@@ -26,6 +26,7 @@
 import org.junit.jupiter.api.Test;
 
 import java.net.ConnectException;
+import java.net.URI;
 
 import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
 import static com.github.tomakehurst.wiremock.client.WireMock.and;
@@ -109,7 +110,7 @@ void shouldTransitionToCompleted_whenControlPlaneRespondCorrectly() {
 
         private DataFlowPrepareMessage createPrepareMessage() {
             return new DataFlowPrepareMessage("any", "any", "any", "any", "dataFlowId", "any", "any",
-                    controlPlane.baseUrl(), "Something-PUSH", emptyList(), emptyMap());
+                    URI.create(controlPlane.baseUrl()), "Something-PUSH", emptyList(), emptyMap());
         }
 
     }
@@ -123,7 +124,7 @@ void shouldRegisterOnTheControlPlane() {
 
             var dataplane = Dataplane.newInstance()
                     .id("dataplane-id")
-                    .endpoint("http://localhost/dataplane")
+                    .endpoint(URI.create("http://localhost/dataplane"))
                     .transferType("SupportedTransferType-PUSH")
                     .label("label-one").label("label-two")
                     .build();
@@ -147,7 +148,7 @@ void shouldFail_whenStatusIsNot200() {
 
             var dataplane = Dataplane.newInstance()
                     .id("dataplane-id")
-                    .endpoint("http://localhost/dataplane")
+                    .endpoint(URI.create("http://localhost/dataplane"))
                     .transferType("SupportedTransferType-PUSH")
                     .label("label-one").label("label-two")
                     .build();
diff --git a/src/test/java/org/eclipse/dataplane/api/ControlPlaneRegistrationApiTest.java b/src/test/java/org/eclipse/dataplane/api/ControlPlaneRegistrationApiTest.java
@@ -26,6 +26,7 @@
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 
+import java.net.URI;
 import java.net.http.HttpRequest;
 import java.util.List;
 import java.util.UUID;
@@ -40,6 +41,7 @@ class ControlPlaneRegistrationApiTest {
             .id("consumer")
             .registerAuthorization(new TestAuthorization())
             .build();
+    private final URI controlPlaneEndpoint = URI.create("http://something");
 
     @BeforeEach
     void setUp() {
@@ -58,7 +60,7 @@ class Register {
         @Test
         void shouldRegisterControlPlane() {
             var controlPlaneId = UUID.randomUUID().toString();
-            var controlPlaneRegistrationMessage = new ControlPlaneRegistrationMessage(controlPlaneId, "http://something");
+            var controlPlaneRegistrationMessage = new ControlPlaneRegistrationMessage(controlPlaneId, controlPlaneEndpoint);
 
             given()
                     .contentType(ContentType.JSON)
@@ -73,13 +75,13 @@ void shouldRegisterControlPlane() {
             var result = sdk.controlPlaneStore().findById(controlPlaneId);
 
             assertThat(result.succeeded());
-            assertThat(result.getContent().getEndpoint()).isEqualTo("http://something");
+            assertThat(result.getContent().getEndpoint()).isEqualTo(controlPlaneEndpoint);
         }
 
         @Test
         void shouldReplaceControlPlane_whenSecondCall() {
             var controlPlaneId = UUID.randomUUID().toString();
-            var controlPlaneRegistrationMessage = new ControlPlaneRegistrationMessage(controlPlaneId, "http://something");
+            var controlPlaneRegistrationMessage = new ControlPlaneRegistrationMessage(controlPlaneId, controlPlaneEndpoint);
 
             given()
                     .contentType(ContentType.JSON)
@@ -91,7 +93,8 @@ void shouldReplaceControlPlane_whenSecondCall() {
                     .log().ifValidationFails()
                     .statusCode(200);
 
-            var updateControlPlane = new ControlPlaneRegistrationMessage(controlPlaneId, "http://new-endpoint");
+            var newControlPlaneEndpoint = URI.create("http://new-endpoint");
+            var updateControlPlane = new ControlPlaneRegistrationMessage(controlPlaneId, newControlPlaneEndpoint);
 
             given()
                     .contentType(ContentType.JSON)
@@ -106,14 +109,14 @@ void shouldReplaceControlPlane_whenSecondCall() {
             var result = sdk.controlPlaneStore().findById(controlPlaneId);
 
             assertThat(result.succeeded());
-            assertThat(result.getContent().getEndpoint()).isEqualTo("http://new-endpoint");
+            assertThat(result.getContent().getEndpoint()).isEqualTo(newControlPlaneEndpoint);
         }
 
         @Test
         void shouldReturnBadRequest_whenRequestedAuthMethodNotSupported() {
             var controlPlaneId = UUID.randomUUID().toString();
             var authorization = new AuthorizationProfile("unsupported");
-            var controlPlaneRegistrationMessage = new ControlPlaneRegistrationMessage(controlPlaneId, "http://something", List.of(authorization));
+            var controlPlaneRegistrationMessage = new ControlPlaneRegistrationMessage(controlPlaneId, controlPlaneEndpoint, List.of(authorization));
 
             given()
                     .contentType(ContentType.JSON)
@@ -131,7 +134,7 @@ void shouldReturnBadRequest_whenRequestedAuthMethodNotSupported() {
         void shouldRegisterAuthorizationType() {
             var controlPlaneId = UUID.randomUUID().toString();
             var authorization = new AuthorizationProfile("token");
-            var controlPlaneRegistrationMessage = new ControlPlaneRegistrationMessage(controlPlaneId, "http://something", List.of(authorization));
+            var controlPlaneRegistrationMessage = new ControlPlaneRegistrationMessage(controlPlaneId, controlPlaneEndpoint, List.of(authorization));
 
             given()
                     .contentType(ContentType.JSON)
@@ -150,7 +153,7 @@ class Delete {
         @Test
         void shouldDeleteControlPlane() {
             var controlPlaneId = UUID.randomUUID().toString();
-            var controlPlaneRegistrationMessage = new ControlPlaneRegistrationMessage(controlPlaneId, "http://something");
+            var controlPlaneRegistrationMessage = new ControlPlaneRegistrationMessage(controlPlaneId, controlPlaneEndpoint);
 
             given()
                     .contentType(ContentType.JSON)
diff --git a/src/test/java/org/eclipse/dataplane/scenario/AuthorizationOauth2Test.java b/src/test/java/org/eclipse/dataplane/scenario/AuthorizationOauth2Test.java
@@ -41,6 +41,7 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.net.URI;
 import java.text.ParseException;
 import java.util.Date;
 import java.util.List;
@@ -130,7 +131,7 @@ private boolean isValidBearerTokenWithSub(String authorization, String controlpl
         }
     }
 
-    private @NonNull DataFlowPrepareMessage createPrepareMessage(String consumerProcessId, String callbackAddress, String transferType) {
+    private @NonNull DataFlowPrepareMessage createPrepareMessage(String consumerProcessId, URI callbackAddress, String transferType) {
         return new DataFlowPrepareMessage("theMessageId", "theParticipantId", "theCounterPartyId",
                 "theDataspaceContext", consumerProcessId, "theAgreementId", "theDatasetId", callbackAddress,
                 transferType, emptyList(), emptyMap());
diff --git a/src/test/java/org/eclipse/dataplane/scenario/AuthorizationTest.java b/src/test/java/org/eclipse/dataplane/scenario/AuthorizationTest.java
@@ -28,6 +28,7 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.net.URI;
 import java.net.http.HttpRequest;
 import java.util.List;
 import java.util.UUID;
@@ -89,7 +90,7 @@ void shouldCommunicateWithControlPlaneUsingRegisteredAuthorization() {
         assertThat(notifyPreparedResult.succeeded()).isTrue();
     }
 
-    private @NonNull DataFlowPrepareMessage createPrepareMessage(String consumerProcessId, String callbackAddress, String transferType) {
+    private @NonNull DataFlowPrepareMessage createPrepareMessage(String consumerProcessId, URI callbackAddress, String transferType) {
         return new DataFlowPrepareMessage("theMessageId", "theParticipantId", "theCounterPartyId",
                 "theDataspaceContext", consumerProcessId, "theAgreementId", "theDatasetId", callbackAddress,
                 transferType, emptyList(), emptyMap());
diff --git a/src/test/java/org/eclipse/dataplane/scenario/ProviderPushTest.java b/src/test/java/org/eclipse/dataplane/scenario/ProviderPushTest.java
diff --git a/src/test/java/org/eclipse/dataplane/scenario/StreamingPushTest.java b/src/test/java/org/eclipse/dataplane/scenario/StreamingPushTest.java
