feat: add trasformers from/to for DataFlowStartMessage

core/data-plane/data-plane-core/src/main/java/org/eclipse/edc/connector/dataplane/framework/iam/DataPlaneAuthorizationServiceImpl.java

@@ -36,7 +36,7 @@ public class DataPlaneAuthorizationServiceImpl implements DataPlaneAuthorization
     public static final String CLAIM_AGREEMENT_ID = "agreement_id";
     public static final String CLAIM_ASSET_ID = "asset_id";
     public static final String CLAIM_PROCESS_ID = "process_id";
-    public static final String CLAIM_TRANSFER_TYPE = "transfer_type";
+    public static final String CLAIM_FLOW_TYPE = "flow_type";
     private final DataPlaneAccessTokenService accessTokenService;
     private final PublicEndpointGeneratorService endpointGenerator;
     private final DataPlaneAccessControlService accessControlService;
@@ -94,7 +94,7 @@ private TokenParameters createTokenParams(DataFlowStartMessage message) {
                 .claims(CLAIM_AGREEMENT_ID, message.getAgreementId())
                 .claims(CLAIM_ASSET_ID, message.getAssetId())
                 .claims(CLAIM_PROCESS_ID, message.getProcessId())
-                .claims(CLAIM_TRANSFER_TYPE, message.getTransferType())
+                .claims(CLAIM_FLOW_TYPE, message.getFlowType().toString())
                 .build();
     }
 

core/data-plane/data-plane-core/src/test/java/org/eclipse/edc/connector/dataplane/framework/iam/DataPlaneAuthorizationServiceImplTest.java

@@ -24,6 +24,7 @@
 import org.eclipse.edc.spi.result.Result;
 import org.eclipse.edc.spi.types.domain.DataAddress;
 import org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage;
+import org.eclipse.edc.spi.types.domain.transfer.FlowType;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentMatchers;
@@ -156,7 +157,7 @@ void authorize_accessNotGranted() {
     private DataFlowStartMessage.Builder createStartMessage() {
         return DataFlowStartMessage.Builder.newInstance()
                 .processId("test-processid")
-                .transferType("test-transfer-type")
+                .flowType(FlowType.PULL)
                 .agreementId("test-agreementid")
                 .participantId("test-participantid")
                 .assetId("test-assetid")

extensions/control-plane/transfer/transfer-data-plane/src/main/java/org/eclipse/edc/connector/transfer/dataplane/flow/ConsumerPullTransferDataFlowController.java

@@ -31,9 +31,9 @@
 
 import static java.lang.String.format;
 import static org.eclipse.edc.connector.transfer.dataplane.spi.TransferDataPlaneConstants.HTTP_PROXY;
-import static org.eclipse.edc.connector.transfer.spi.flow.FlowType.PULL;
 import static org.eclipse.edc.spi.response.ResponseStatus.FATAL_ERROR;
 import static org.eclipse.edc.spi.response.StatusResult.failure;
+import static org.eclipse.edc.spi.types.domain.transfer.FlowType.PULL;
 
 public class ConsumerPullTransferDataFlowController implements DataFlowController {
 

extensions/control-plane/transfer/transfer-data-plane/src/main/java/org/eclipse/edc/connector/transfer/dataplane/flow/ProviderPushTransferDataFlowController.java

@@ -36,8 +36,8 @@
 
 import static java.util.stream.Collectors.toSet;
 import static org.eclipse.edc.connector.transfer.dataplane.spi.TransferDataPlaneConstants.HTTP_PROXY;
-import static org.eclipse.edc.connector.transfer.spi.flow.FlowType.PULL;
-import static org.eclipse.edc.connector.transfer.spi.flow.FlowType.PUSH;
+import static org.eclipse.edc.spi.types.domain.transfer.FlowType.PULL;
+import static org.eclipse.edc.spi.types.domain.transfer.FlowType.PUSH;
 
 public class ProviderPushTransferDataFlowController implements DataFlowController {
 
@@ -68,7 +68,7 @@ public boolean canHandle(TransferProcess transferProcess) {
                 .processId(transferProcess.getId())
                 .sourceDataAddress(transferProcess.getContentDataAddress())
                 .destinationDataAddress(transferProcess.getDataDestination())
-                .transferType(transferProcess.getTransferType())
+                .flowType(PUSH)
                 .callbackAddress(callbackUrl != null ? callbackUrl.get() : null)
                 .build();
 

extensions/data-plane/data-plane-signaling/data-plane-signaling-transform/src/main/java/org/eclipse/edc/connector/api/signaling/transform/from/JsonObjectFromDataFlowStartMessageTransformer.java

@@ -0,0 +1,68 @@
+/*
+ *  Copyright (c) 2024 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - initial API and implementation
+ *
+ */
+
+package org.eclipse.edc.connector.api.signaling.transform.from;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import jakarta.json.JsonBuilderFactory;
+import jakarta.json.JsonObject;
+import org.eclipse.edc.jsonld.spi.transformer.AbstractJsonLdTransformer;
+import org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage;
+import org.eclipse.edc.transform.spi.TransformerContext;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import static org.eclipse.edc.jsonld.spi.JsonLdKeywords.TYPE;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.DC_DATA_FLOW_START_MESSAGE_PROCESS_ID;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_AGREEMENT_ID;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_DATASET_ID;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_DESTINATION_CALLBACK_ADDRESS;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_DESTINATION_DATA_ADDRESS;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_FLOW_TYPE;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_PARTICIPANT_ID;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_PROPERTIES;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_SOURCE_DATA_ADDRESS;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_TYPE;
+
+/**
+ * Converts from a {@link DataFlowStartMessage} to a {@link JsonObject} in JSON-LD expanded form .
+ */
+public class JsonObjectFromDataFlowStartMessageTransformer extends AbstractJsonLdTransformer<DataFlowStartMessage, JsonObject> {
+    private final JsonBuilderFactory jsonFactory;
+    private final ObjectMapper mapper;
+
+    public JsonObjectFromDataFlowStartMessageTransformer(JsonBuilderFactory jsonFactory, ObjectMapper mapper) {
+        super(DataFlowStartMessage.class, JsonObject.class);
+        this.jsonFactory = jsonFactory;
+        this.mapper = mapper;
+    }
+
+    @Override
+    public @Nullable JsonObject transform(@NotNull DataFlowStartMessage message, @NotNull TransformerContext context) {
+        var propertiesBuilder = jsonFactory.createObjectBuilder();
+        transformProperties(message.getProperties(), propertiesBuilder, mapper, context);
+        return jsonFactory.createObjectBuilder()
+                .add(TYPE, EDC_DATA_FLOW_START_MESSAGE_TYPE)
+                .add(EDC_DATA_FLOW_START_MESSAGE_FLOW_TYPE, message.getFlowType().toString())
+                .add(EDC_DATA_FLOW_START_MESSAGE_AGREEMENT_ID, message.getAgreementId())
+                .add(DC_DATA_FLOW_START_MESSAGE_PROCESS_ID, message.getProcessId())
+                .add(EDC_DATA_FLOW_START_MESSAGE_DATASET_ID, message.getAssetId())
+                .add(EDC_DATA_FLOW_START_MESSAGE_PROPERTIES, propertiesBuilder)
+                .add(EDC_DATA_FLOW_START_MESSAGE_DESTINATION_CALLBACK_ADDRESS, message.getCallbackAddress().toString())
+                .add(EDC_DATA_FLOW_START_MESSAGE_DESTINATION_DATA_ADDRESS, context.transform(message.getDestinationDataAddress(), JsonObject.class))
+                .add(EDC_DATA_FLOW_START_MESSAGE_SOURCE_DATA_ADDRESS, context.transform(message.getSourceDataAddress(), JsonObject.class))
+                .add(EDC_DATA_FLOW_START_MESSAGE_PARTICIPANT_ID, message.getParticipantId())
+                .build();
+    }
+}

extensions/data-plane/data-plane-signaling/data-plane-signaling-transform/src/main/java/org/eclipse/edc/connector/api/signaling/transform/to/JsonObjectToDataFlowStartMessageTransformer.java

@@ -0,0 +1,82 @@
+/*
+ *  Copyright (c) 2024 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - initial API and implementation
+ *
+ */
+
+package org.eclipse.edc.connector.api.signaling.transform.to;
+
+import jakarta.json.JsonObject;
+import jakarta.json.JsonValue;
+import org.eclipse.edc.jsonld.spi.transformer.AbstractJsonLdTransformer;
+import org.eclipse.edc.spi.types.domain.DataAddress;
+import org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage;
+import org.eclipse.edc.spi.types.domain.transfer.FlowType;
+import org.eclipse.edc.transform.spi.TransformerContext;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import java.net.URI;
+import java.util.Optional;
+
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.Builder;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.DC_DATA_FLOW_START_MESSAGE_PROCESS_ID;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_AGREEMENT_ID;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_DATASET_ID;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_DESTINATION_CALLBACK_ADDRESS;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_DESTINATION_DATA_ADDRESS;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_FLOW_TYPE;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_PARTICIPANT_ID;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_PROPERTIES;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_SOURCE_DATA_ADDRESS;
+
+/**
+ * Converts from a {@link JsonObject} in JSON-LD expanded form to a {@link DataFlowStartMessage}.
+ */
+public class JsonObjectToDataFlowStartMessageTransformer extends AbstractJsonLdTransformer<JsonObject, DataFlowStartMessage> {
+
+    public JsonObjectToDataFlowStartMessageTransformer() {
+        super(JsonObject.class, DataFlowStartMessage.class);
+    }
+
+    @Override
+    public @Nullable DataFlowStartMessage transform(@NotNull JsonObject object, @NotNull TransformerContext context) {
+        var builder = Builder.newInstance();
+        visitProperties(object, (s, jsonValue) -> transformProperties(s, jsonValue, builder, context));
+        return builder.build();
+    }
+
+    private void transformProperties(String key, JsonValue jsonValue, Builder builder, TransformerContext context) {
+        switch (key) {
+            case DC_DATA_FLOW_START_MESSAGE_PROCESS_ID -> builder.processId(transformString(jsonValue, context));
+            case EDC_DATA_FLOW_START_MESSAGE_AGREEMENT_ID -> builder.agreementId(transformString(jsonValue, context));
+            case EDC_DATA_FLOW_START_MESSAGE_DATASET_ID -> builder.assetId(transformString(jsonValue, context));
+            case EDC_DATA_FLOW_START_MESSAGE_DESTINATION_CALLBACK_ADDRESS ->
+                    Optional.ofNullable(transformString(jsonValue, context)).map(URI::create).ifPresent(builder::callbackAddress);
+            case EDC_DATA_FLOW_START_MESSAGE_PROPERTIES -> {
+                var props = jsonValue.asJsonArray().getJsonObject(0);
+                visitProperties(props, (k, val) -> transformProperties(k, val, builder, context));
+            }
+            case EDC_DATA_FLOW_START_MESSAGE_PARTICIPANT_ID ->
+                    builder.participantId(transformString(jsonValue, context));
+            case EDC_DATA_FLOW_START_MESSAGE_FLOW_TYPE ->
+                    builder.flowType(FlowType.valueOf(transformString(jsonValue, context)));
+
+            case EDC_DATA_FLOW_START_MESSAGE_DESTINATION_DATA_ADDRESS ->
+                    builder.destinationDataAddress(transformObject(jsonValue, DataAddress.class, context));
+
+            case EDC_DATA_FLOW_START_MESSAGE_SOURCE_DATA_ADDRESS ->
+                    builder.sourceDataAddress(transformObject(jsonValue, DataAddress.class, context));
+
+            default -> builder.property(key, transformString(jsonValue, context));
+        }
+    }
+}

extensions/data-plane/data-plane-signaling/data-plane-signaling-transform/src/test/java/org/eclipse/edc/connector/api/signaling/transform/from/JsonObjectFromDataFlowStartMessageTransformerTest.java

@@ -0,0 +1,86 @@
+/*
+ *  Copyright (c) 2024 Bayerische Motoren Werke Aktiengesellschaft (BMW AG)
+ *
+ *  This program and the accompanying materials are made available under the
+ *  terms of the Apache License, Version 2.0 which is available at
+ *  https://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  SPDX-License-Identifier: Apache-2.0
+ *
+ *  Contributors:
+ *       Bayerische Motoren Werke Aktiengesellschaft (BMW AG) - initial API and implementation
+ *
+ */
+
+package org.eclipse.edc.connector.api.signaling.transform.from;
+
+import jakarta.json.Json;
+import org.eclipse.edc.spi.types.domain.DataAddress;
+import org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage;
+import org.eclipse.edc.spi.types.domain.transfer.FlowType;
+import org.eclipse.edc.transform.spi.TransformerContext;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.net.URI;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.eclipse.edc.jsonld.spi.JsonLdKeywords.TYPE;
+import static org.eclipse.edc.jsonld.util.JacksonJsonLd.createObjectMapper;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.DC_DATA_FLOW_START_MESSAGE_PROCESS_ID;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_AGREEMENT_ID;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_DATASET_ID;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_DESTINATION_CALLBACK_ADDRESS;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_DESTINATION_DATA_ADDRESS;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_FLOW_TYPE;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_PARTICIPANT_ID;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_SOURCE_DATA_ADDRESS;
+import static org.eclipse.edc.spi.types.domain.transfer.DataFlowStartMessage.EDC_DATA_FLOW_START_MESSAGE_TYPE;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.isA;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class JsonObjectFromDataFlowStartMessageTransformerTest {
+
+    private final TransformerContext context = mock(TransformerContext.class);
+    private JsonObjectFromDataFlowStartMessageTransformer transformer;
+
+    @BeforeEach
+    void setUp() {
+        transformer = new JsonObjectFromDataFlowStartMessageTransformer(Json.createBuilderFactory(Map.of()), createObjectMapper());
+        when(context.transform(isA(DataAddress.class), any())).thenReturn(Json.createObjectBuilder().build());
+    }
+
+    @Test
+    void transform() {
+
+        var message = DataFlowStartMessage.Builder.newInstance()
+                .processId("processId")
+                .assetId("assetId")
+                .agreementId("agreementId")
+                .participantId("participantId")
+                .flowType(FlowType.PUSH)
+                .callbackAddress(URI.create("http://localhost"))
+                .sourceDataAddress(DataAddress.Builder.newInstance().type("sourceType").build())
+                .destinationDataAddress(DataAddress.Builder.newInstance().type("destType").build())
+                .build();
+
+        var jsonObject = transformer.transform(message, context);
+
+        assertThat(jsonObject).isNotNull();
+
+        assertThat(jsonObject.getJsonString(TYPE).getString()).isEqualTo(EDC_DATA_FLOW_START_MESSAGE_TYPE);
+        assertThat(jsonObject.getJsonString(DC_DATA_FLOW_START_MESSAGE_PROCESS_ID).getString()).isEqualTo(message.getProcessId());
+        assertThat(jsonObject.getJsonString(EDC_DATA_FLOW_START_MESSAGE_DATASET_ID).getString()).isEqualTo(message.getAssetId());
+        assertThat(jsonObject.getJsonString(EDC_DATA_FLOW_START_MESSAGE_AGREEMENT_ID).getString()).isEqualTo(message.getAgreementId());
+        assertThat(jsonObject.getJsonString(EDC_DATA_FLOW_START_MESSAGE_PARTICIPANT_ID).getString()).isEqualTo(message.getParticipantId());
+        assertThat(jsonObject.getJsonString(EDC_DATA_FLOW_START_MESSAGE_DESTINATION_CALLBACK_ADDRESS).getString()).isEqualTo(message.getCallbackAddress().toString());
+        assertThat(jsonObject.getJsonString(EDC_DATA_FLOW_START_MESSAGE_FLOW_TYPE).getString()).isEqualTo(message.getFlowType().toString());
+        assertThat(jsonObject.get(EDC_DATA_FLOW_START_MESSAGE_DESTINATION_DATA_ADDRESS)).isNotNull();
+        assertThat(jsonObject.get(EDC_DATA_FLOW_START_MESSAGE_SOURCE_DATA_ADDRESS)).isNotNull();
+
+    }
+
+}