chore: handle missing destination in protocol service

eclipse-edc · Jul 20, 2023 · b5bf5b2 · b5bf5b2
1 parent b62b87f
commit b5bf5b2
Show file tree

Hide file tree

Showing 2 changed files with 39 additions and 5 deletions.
diff --git a/...org/eclipse/edc/connector/service/transferprocess/TransferProcessProtocolServiceImpl.java b/...org/eclipse/edc/connector/service/transferprocess/TransferProcessProtocolServiceImpl.java
@@ -37,6 +37,7 @@
 import org.eclipse.edc.spi.monitor.Monitor;
 import org.eclipse.edc.spi.result.Result;
 import org.eclipse.edc.spi.telemetry.Telemetry;
+import org.eclipse.edc.spi.types.domain.DataAddress;
 import org.eclipse.edc.transaction.spi.TransactionContext;
 import org.jetbrains.annotations.NotNull;
 
@@ -46,6 +47,7 @@
 
 import static java.lang.String.format;
 import static java.util.UUID.randomUUID;
+import static org.eclipse.edc.connector.transfer.dataplane.spi.TransferDataPlaneConstants.HTTP_PROXY;
 import static org.eclipse.edc.connector.transfer.spi.types.TransferProcess.Type.CONSUMER;
 import static org.eclipse.edc.connector.transfer.spi.types.TransferProcess.Type.PROVIDER;
 
@@ -86,9 +88,12 @@ public ServiceResult<TransferProcess> notifyRequested(TransferRequestMessage mes
             return ServiceResult.badRequest("ContractId is not valid: " + contractIdResult.getFailureDetail());
         }
 
-        var validDestination = dataAddressValidator.validate(message.getDataDestination());
-        if (validDestination.failed()) {
-            return ServiceResult.badRequest(validDestination.getFailureMessages());
+        var destination = message.getDataDestination();
+        if (destination != null) { // NOTE: optional property according to protocol spec
+            var validDestination = dataAddressValidator.validate(destination);
+            if (validDestination.failed()) {
+                return ServiceResult.badRequest(validDestination.getFailureMessages());
+            }
         }
 
         return transactionContext.execute(() ->
@@ -132,12 +137,15 @@ public ServiceResult<TransferProcess> findById(String id, ClaimToken claimToken)
     @NotNull
     private ServiceResult<TransferProcess> requestedAction(TransferRequestMessage message, ContractId contractId) {
         var assetId = contractId.assetIdPart();
-
+
+        // NOTE: if destination is missing a pull transfer is requested
+        var destination = message.getDataDestination() != null ? message.getDataDestination() :
+                DataAddress.Builder.newInstance().type(HTTP_PROXY).build();
         var dataRequest = DataRequest.Builder.newInstance()
                 .id(message.getProcessId())
                 .protocol(message.getProtocol())
                 .connectorAddress(message.getCallbackAddress())
-                .dataDestination(message.getDataDestination())
+                .dataDestination(destination)
                 .assetId(assetId)
                 .contractId(message.getContractId())
                 .build();

diff --git a/...eclipse/edc/connector/service/transferprocess/TransferProcessProtocolServiceImplTest.java b/...eclipse/edc/connector/service/transferprocess/TransferProcessProtocolServiceImplTest.java
@@ -57,6 +57,7 @@
 import java.util.stream.Stream;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.eclipse.edc.connector.transfer.dataplane.spi.TransferDataPlaneConstants.HTTP_PROXY;
 import static org.eclipse.edc.connector.transfer.spi.types.TransferProcessStates.COMPLETED;
 import static org.eclipse.edc.connector.transfer.spi.types.TransferProcessStates.INITIAL;
 import static org.eclipse.edc.connector.transfer.spi.types.TransferProcessStates.REQUESTED;
@@ -194,6 +195,31 @@ void notifyRequested_invalidDestination_shouldNotInitiateTransfer() {
         verify(store, never()).updateOrCreate(any());
         verifyNoInteractions(listener);
     }
+
+    @Test
+    void notifyRequested_missingDestination_shouldInitiateTransfer() {
+        var message = TransferRequestMessage.Builder.newInstance()
+                .processId("transferProcessId")
+                .protocol("protocol")
+                .contractId(ContractId.create("definitionId", "assetId").toString())
+                .callbackAddress("http://any")
+                .build();
+        when(negotiationStore.findContractAgreement(any())).thenReturn(contractAgreement());
+        when(validationService.validateAgreement(any(), any())).thenReturn(Result.success(null));
+
+        var result = service.notifyRequested(message, claimToken());
+
+        assertThat(result).isSucceeded().satisfies(tp -> {
+            assertThat(tp.getCorrelationId()).isEqualTo("transferProcessId");
+            assertThat(tp.getConnectorAddress()).isEqualTo("http://any");
+            assertThat(tp.getAssetId()).isEqualTo("assetId");
+            assertThat(tp.getDataDestination().getType()).isEqualTo(HTTP_PROXY);
+        });
+        verify(listener).preCreated(any());
+        verify(store).updateOrCreate(argThat(t -> t.getState() == INITIAL.code()));
+        verify(listener).initiated(any());
+        verify(transactionContext, atLeastOnce()).execute(any(TransactionContext.ResultTransactionBlock.class));
+    }
 
     @Test
     void notifyStarted_shouldTransitionToStarted() {