feat: prioritized transfer services (#4876)

Author: ronjaquensel

core/data-plane/data-plane-core/src/main/java/org/eclipse/edc/connector/dataplane/framework/registry/TransferServiceRegistryImpl.java

@@ -9,6 +9,7 @@
  *
  *  Contributors:
  *       Microsoft Corporation - initial API and implementation
+ *       Cofinity-X - prioritized transfer services
  *
  */
 
@@ -20,14 +21,15 @@
 import org.jetbrains.annotations.Nullable;
 
 import java.util.Collection;
+import java.util.Comparator;
 import java.util.LinkedHashSet;
 
 /**
  * Default {@link TransferServiceRegistry} implementation.
  */
 public class TransferServiceRegistryImpl implements TransferServiceRegistry {
 
-    private final Collection<TransferService> transferServices = new LinkedHashSet<>();
+    private final Collection<PrioritizedTransferService> transferServices = new LinkedHashSet<>();
     private final TransferServiceSelectionStrategy transferServiceSelectionStrategy;
 
     public TransferServiceRegistryImpl(TransferServiceSelectionStrategy transferServiceSelectionStrategy) {
@@ -36,13 +38,34 @@ public TransferServiceRegistryImpl(TransferServiceSelectionStrategy transferServ
 
     @Override
     public void registerTransferService(TransferService transferService) {
-        transferServices.add(transferService);
+        transferServices.add(new PrioritizedTransferService(0, transferService));
     }
-
+    
+    @Override
+    public void registerTransferService(int priority, TransferService transferService) {
+        transferServices.add(new PrioritizedTransferService(priority, transferService));
+    }
+    
     @Override
     @Nullable
     public TransferService resolveTransferService(DataFlowStartMessage request) {
-        var possibleServices = transferServices.stream().filter(s -> s.canHandle(request));
+        var prioritizedServicesPresent = transferServices.stream()
+                .map(PrioritizedTransferService::priority)
+                .anyMatch(priority -> priority > 0);
+
+        if (prioritizedServicesPresent) {
+            return transferServices.stream()
+                    .filter(pts -> pts.service.canHandle(request))
+                    .sorted(Comparator.comparingInt(pts -> -pts.priority))
+                    .map(PrioritizedTransferService::service)
+                    .findFirst().orElse(null);
+        }
+
+        var possibleServices = transferServices.stream()
+                .map(PrioritizedTransferService::service)
+                .filter(ts -> ts.canHandle(request));
         return transferServiceSelectionStrategy.chooseTransferService(request, possibleServices);
     }
+    
+    record PrioritizedTransferService(int priority, TransferService service) { }
 }

core/data-plane/data-plane-core/src/main/java/org/eclipse/edc/connector/dataplane/framework/registry/TransferServiceSelectionStrategy.java

@@ -9,6 +9,7 @@
  *
  *  Contributors:
  *       Microsoft Corporation - Initial implementation
+ *       Cofinity-X - prioritized transfer services
  *
  */
 
@@ -23,7 +24,11 @@
 /**
  * Functional interface for selecting which of (potentially) multiple {@link TransferService}s to use
  * for serving a particular {@link DataFlowStartMessage}.
+ *
+ * @deprecated use transfer service prioritization, see
+ *      {@link org.eclipse.edc.connector.dataplane.spi.registry.TransferServiceRegistry#registerTransferService(int, TransferService)}.
  */
+@Deprecated(since = "0.12.0", forRemoval = true)
 public interface TransferServiceSelectionStrategy {
     /**
      * Selects which of (potentially) multiple {@link TransferService}s to use

core/data-plane/data-plane-core/src/test/java/org/eclipse/edc/connector/dataplane/framework/registry/TransferServiceRegistryImplTest.java

@@ -9,6 +9,7 @@
  *
  *  Contributors:
  *       Microsoft Corporation - initial API and implementation
+ *       Cofinity-X - prioritized transfer services
  *
  */
 
@@ -23,70 +24,100 @@
 import java.util.stream.Stream;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.eq;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 class TransferServiceRegistryImplTest {
-    TransferService transferService = mock(TransferService.class);
-    TransferService transferService2 = mock(TransferService.class);
-
-    DataFlowStartMessage request = createRequest().build();
-    TransferServiceSelectionStrategy transferServiceSelectionStrategy = mock(TransferServiceSelectionStrategy.class);
+    private TransferService transferService = mock(TransferService.class);
+    private TransferService transferService2 = mock(TransferService.class);
+    
+    private DataFlowStartMessage request = createRequest().build();
+    private TransferServiceSelectionStrategy transferServiceSelectionStrategy = mock(TransferServiceSelectionStrategy.class);
     @SuppressWarnings("unchecked")
-    ArgumentCaptor<Stream<TransferService>> streamCaptor = ArgumentCaptor.forClass(Stream.class);
-
+    private ArgumentCaptor<Stream<TransferService>> streamCaptor = ArgumentCaptor.forClass(Stream.class);
+    
+    private TransferServiceRegistryImpl registry = new TransferServiceRegistryImpl(transferServiceSelectionStrategy);
+    
     @Test
-    void resolveTransferService_filters_matches() {
+    void resolveTransferService_noServicesRegistered_shouldReturnNull() {
+        var service = registry.resolveTransferService(request);
+        
+        assertThat(service).isNull();
+    }
+    
+    @Test
+    void resolveTransferService_noServiceCanHandle_shouldReturnNull() {
+        registry.registerTransferService(transferService);
         when(transferService.canHandle(request)).thenReturn(false);
+        
+        var service = registry.resolveTransferService(request);
+        
+        assertThat(service).isNull();
+    }
+    
+    @Test
+    void resolveTransferService_withPriorities_shouldReturnHighestPriorityService() {
+        registry.registerTransferService(transferService);
+        registry.registerTransferService(1, transferService2);
+        when(transferService.canHandle(request)).thenReturn(true);
         when(transferService2.canHandle(request)).thenReturn(true);
-
-        createRegistryAndResolveForRequest();
-
-        assertThat(streamCaptor.getValue()).containsExactly(transferService2);
+        
+        var service = registry.resolveTransferService(request);
+        
+        assertThat(service).isEqualTo(transferService2);
     }
-
+    
     @Test
-    void resolveTransferService_handles_multipleMatches() {
+    void resolveTransferService_withSamePriority_shouldReturnFirstWithHighestPriority() {
+        registry.registerTransferService(1, transferService);
+        registry.registerTransferService(1, transferService2);
         when(transferService.canHandle(request)).thenReturn(true);
         when(transferService2.canHandle(request)).thenReturn(true);
-
-        createRegistryAndResolveForRequest();
-
-        assertThat(streamCaptor.getValue()).containsExactly(transferService, transferService2);
+        
+        var service = registry.resolveTransferService(request);
+        
+        assertThat(service).isEqualTo(transferService);
     }
-
+    
     @Test
-    void resolveTransferService_handles_noMatch() {
-        var registry = new TransferServiceRegistryImpl(transferServiceSelectionStrategy);
-
+    void resolveTransferService_noPriorityAndNoneCanHandle_shouldApplyStrategyWithEmptyStream() {
+        registry.registerTransferService(transferService);
+        registry.registerTransferService(transferService2);
+        when(transferService.canHandle(request)).thenReturn(false);
+        when(transferService2.canHandle(request)).thenReturn(false);
+        
         registry.resolveTransferService(request);
-
+        
         verify(transferServiceSelectionStrategy).chooseTransferService(eq(request), streamCaptor.capture());
         assertThat(streamCaptor.getValue()).isEmpty();
     }
-
+    
     @Test
-    void resolveTransferService_returns_strategyResult() {
-        var registry = new TransferServiceRegistryImpl(transferServiceSelectionStrategy);
-        when(transferServiceSelectionStrategy.chooseTransferService(eq(request), any()))
-                .thenReturn(transferService);
-
-        var resolved = registry.resolveTransferService(request);
-
-        assertThat(resolved).isSameAs(transferService);
+    void resolveTransferService_noPriorityAndOneCanHandle_shouldApplyStrategyWithOneService() {
+        registry.registerTransferService(transferService);
+        registry.registerTransferService(transferService2);
+        when(transferService.canHandle(request)).thenReturn(true);
+        when(transferService2.canHandle(request)).thenReturn(false);
+        
+        registry.resolveTransferService(request);
+        
+        verify(transferServiceSelectionStrategy).chooseTransferService(eq(request), streamCaptor.capture());
+        assertThat(streamCaptor.getValue()).containsExactly(transferService);
     }
-
-    private void createRegistryAndResolveForRequest() {
-        var registry = new TransferServiceRegistryImpl(transferServiceSelectionStrategy);
+    
+    @Test
+    void resolveTransferService_noPriorityAndAllCanHandle_shouldApplyStrategyWithAllServices() {
         registry.registerTransferService(transferService);
         registry.registerTransferService(transferService2);
-
+        when(transferService.canHandle(request)).thenReturn(true);
+        when(transferService2.canHandle(request)).thenReturn(true);
+        
         registry.resolveTransferService(request);
-
+        
         verify(transferServiceSelectionStrategy).chooseTransferService(eq(request), streamCaptor.capture());
+        assertThat(streamCaptor.getValue()).containsExactly(transferService, transferService2);
     }
 
     private DataFlowStartMessage.Builder createRequest() {

spi/data-plane/data-plane-spi/src/main/java/org/eclipse/edc/connector/dataplane/spi/registry/TransferServiceRegistry.java

@@ -9,6 +9,7 @@
  *
  *  Contributors:
  *       Microsoft Corporation - initial API and implementation
+ *       Cofinity-X - prioritized transfer services
  *
  */
 
@@ -24,13 +25,23 @@
  */
 @ExtensionPoint
 public interface TransferServiceRegistry {
+    
     /**
      * Adds a {@link TransferService} to the collection of services that can perform data transfers.
+     * The priority is set to 0.
      *
      * @param transferService the service to add.
      */
     void registerTransferService(TransferService transferService);
-
+    
+    /**
+     * Adds a {@link TransferService} with given priority to the collection of services that can
+     * perform data transfers. Higher priorities will be preferred during selection.
+     *
+     * @param priority the priority
+     * @param transferService the service to add.
+     */
+    void registerTransferService(int priority, TransferService transferService);
 
     /**
      * Resolves a {@link TransferService}s to use for serving a particular {@link DataFlowStartMessage}.