feat: implement Asset/Dataset specialization pt. 2

core/control-plane/control-plane-catalog/build.gradle.kts

@@ -22,7 +22,8 @@ dependencies {
     api(project(":spi:control-plane:transfer-spi"))
     api(project(":spi:control-plane:asset-spi"))
 
-    testImplementation(project(":tests:junit-base"));
+    implementation(project(":spi:common:data-address:data-address-http-data-spi"))
+    testImplementation(project(":tests:junit-base"))
 
     testImplementation(project(":core:common:connector-core"))
     testImplementation(project(":core:control-plane:control-plane-core"))

core/control-plane/control-plane-catalog/src/main/java/org/eclipse/edc/connector/controlplane/catalog/DatasetResolverImpl.java

@@ -16,19 +16,23 @@
 
 import org.eclipse.edc.connector.controlplane.asset.spi.domain.Asset;
 import org.eclipse.edc.connector.controlplane.asset.spi.index.AssetIndex;
+import org.eclipse.edc.connector.controlplane.catalog.spi.Catalog;
+import org.eclipse.edc.connector.controlplane.catalog.spi.DataService;
 import org.eclipse.edc.connector.controlplane.catalog.spi.Dataset;
 import org.eclipse.edc.connector.controlplane.catalog.spi.DatasetResolver;
 import org.eclipse.edc.connector.controlplane.catalog.spi.DistributionResolver;
 import org.eclipse.edc.connector.controlplane.contract.spi.ContractOfferId;
 import org.eclipse.edc.connector.controlplane.contract.spi.offer.ContractDefinitionResolver;
 import org.eclipse.edc.connector.controlplane.contract.spi.types.offer.ContractDefinition;
 import org.eclipse.edc.connector.controlplane.policy.spi.store.PolicyDefinitionStore;
+import org.eclipse.edc.dataaddress.httpdata.spi.HttpDataAddressSchema;
 import org.eclipse.edc.policy.model.PolicyType;
 import org.eclipse.edc.spi.agent.ParticipantAgent;
 import org.eclipse.edc.spi.query.CriterionOperatorRegistry;
 import org.eclipse.edc.spi.query.QuerySpec;
 import org.jetbrains.annotations.NotNull;
 
+import java.util.Base64;
 import java.util.List;
 import java.util.Optional;
 import java.util.function.Predicate;
@@ -75,10 +79,23 @@ public Dataset getById(ParticipantAgent agent, String id) {
                 .orElse(null);
     }
 
+    private Dataset.Builder buildDataset(Asset asset) {
+        if (!asset.isCatalog()) {
+            return Dataset.Builder.newInstance();
+        }
+
+        return Catalog.Builder.newInstance()
+                .dataService(DataService.Builder.newInstance()
+                        .id(Base64.getUrlEncoder().encodeToString(asset.getId().getBytes()))
+                        .endpointDescription(asset.getDescription())
+                        .endpointUrl(asset.getDataAddress().getStringProperty(HttpDataAddressSchema.BASE_URL, null))
+                        .build());
+    }
+
     private Dataset toDataset(List<ContractDefinition> contractDefinitions, Asset asset) {
 
         var distributions = distributionResolver.getDistributions(asset);
-        var datasetBuilder = Dataset.Builder.newInstance()
+        var datasetBuilder = buildDataset(asset)
                 .id(asset.getId())
                 .distributions(distributions)
                 .properties(asset.getProperties());

core/control-plane/control-plane-catalog/src/main/java/org/eclipse/edc/connector/controlplane/catalog/DefaultDistributionResolver.java

@@ -16,11 +16,13 @@
 package org.eclipse.edc.connector.controlplane.catalog;
 
 import org.eclipse.edc.connector.controlplane.asset.spi.domain.Asset;
+import org.eclipse.edc.connector.controlplane.catalog.spi.DataService;
 import org.eclipse.edc.connector.controlplane.catalog.spi.DataServiceRegistry;
 import org.eclipse.edc.connector.controlplane.catalog.spi.Distribution;
 import org.eclipse.edc.connector.controlplane.catalog.spi.DistributionResolver;
 import org.eclipse.edc.connector.controlplane.transfer.spi.flow.DataFlowManager;
 
+import java.util.Base64;
 import java.util.List;
 
 public class DefaultDistributionResolver implements DistributionResolver {
@@ -35,6 +37,14 @@ public DefaultDistributionResolver(DataServiceRegistry dataServiceRegistry, Data
 
     @Override
     public List<Distribution> getDistributions(Asset asset) {
+        if (asset.isCatalog()) {
+            return List.of(Distribution.Builder.newInstance()
+                    .format(asset.getDataAddress().getType())
+                    .dataService(DataService.Builder.newInstance()
+                            .id(Base64.getUrlEncoder().encodeToString(asset.getId().getBytes()))
+                            .build())
+                    .build());
+        }
         return dataFlowManager.transferTypesFor(asset).stream().map(this::createDistribution).toList();
     }
 

core/control-plane/control-plane-catalog/src/test/java/org/eclipse/edc/connector/controlplane/catalog/DatasetResolverImplTest.java

@@ -18,6 +18,7 @@
 import org.assertj.core.api.iterable.ThrowingExtractor;
 import org.eclipse.edc.connector.controlplane.asset.spi.domain.Asset;
 import org.eclipse.edc.connector.controlplane.asset.spi.index.AssetIndex;
+import org.eclipse.edc.connector.controlplane.catalog.spi.Catalog;
 import org.eclipse.edc.connector.controlplane.catalog.spi.DataService;
 import org.eclipse.edc.connector.controlplane.catalog.spi.Dataset;
 import org.eclipse.edc.connector.controlplane.catalog.spi.DatasetResolver;
@@ -28,12 +29,14 @@
 import org.eclipse.edc.connector.controlplane.contract.spi.types.offer.ContractDefinition;
 import org.eclipse.edc.connector.controlplane.policy.spi.PolicyDefinition;
 import org.eclipse.edc.connector.controlplane.policy.spi.store.PolicyDefinitionStore;
+import org.eclipse.edc.dataaddress.httpdata.spi.HttpDataAddressSchema;
 import org.eclipse.edc.policy.model.Policy;
 import org.eclipse.edc.query.CriterionOperatorRegistryImpl;
 import org.eclipse.edc.spi.agent.ParticipantAgent;
 import org.eclipse.edc.spi.message.Range;
 import org.eclipse.edc.spi.query.Criterion;
 import org.eclipse.edc.spi.query.QuerySpec;
+import org.eclipse.edc.spi.types.domain.DataAddress;
 import org.jetbrains.annotations.NotNull;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -219,6 +222,37 @@ void query_shouldLimitDataset_whenMultipleDefinitionsWithSameAssets() {
                 .map(getId()).containsExactly("6", "7");
     }
 
+    @Test
+    void query_shouldReturnCatalogWithinCatalog_whenAssetIsCatalogAsset() {
+        var contractDefinition = contractDefinitionBuilder("definitionId").contractPolicyId("contractPolicyId").build();
+        var contractPolicy = Policy.Builder.newInstance().build();
+        var distribution = Distribution.Builder.newInstance().dataService(DataService.Builder.newInstance()
+                        .endpointDescription("test-asset-desc")
+                        .endpointUrl("https://foo.bar/baz")
+                        .build())
+                .format(HttpDataAddressSchema.HTTP_DATA_TYPE).build();
+
+        when(contractDefinitionResolver.definitionsFor(any())).thenReturn(Stream.of(contractDefinition));
+        when(assetIndex.queryAssets(isA(QuerySpec.class))).thenReturn(Stream.of(createAsset("assetId")
+                .property(Asset.PROPERTY_IS_CATALOG, true)
+                .dataAddress(DataAddress.Builder.newInstance().type(HttpDataAddressSchema.HTTP_DATA_TYPE).build())
+                .build()));
+        when(policyStore.findById("contractPolicyId")).thenReturn(PolicyDefinition.Builder.newInstance().policy(contractPolicy).build());
+        when(distributionResolver.getDistributions(isA(Asset.class))).thenReturn(List.of(distribution));
+
+        var datasets = datasetResolver.query(createParticipantAgent(), QuerySpec.none());
+
+        assertThat(datasets).isNotNull().hasSize(1).first().satisfies(dataset -> {
+            assertThat(dataset).isInstanceOf(Catalog.class);
+            assertThat(dataset.getId()).isEqualTo("assetId");
+            assertThat(dataset.getOffers()).hasSize(1).allSatisfy((id, policy) -> {
+                assertThat(ContractOfferId.parseId(id)).isSucceeded().extracting(ContractOfferId::definitionPart).asString().isEqualTo("definitionId");
+                assertThat(policy.getType()).isEqualTo(OFFER);
+                assertThat(policy.getTarget()).isEqualTo(null);
+            });
+        });
+    }
+
     @Test
     void getById_shouldReturnDataset() {
         var policy1 = Policy.Builder.newInstance().inheritsFrom("inherits1").build();

core/control-plane/control-plane-catalog/src/test/java/org/eclipse/edc/connector/controlplane/catalog/DefaultDistributionResolverTest.java

@@ -18,6 +18,7 @@
 import org.eclipse.edc.connector.controlplane.catalog.spi.DataService;
 import org.eclipse.edc.connector.controlplane.catalog.spi.DataServiceRegistry;
 import org.eclipse.edc.connector.controlplane.transfer.spi.flow.DataFlowManager;
+import org.eclipse.edc.dataaddress.httpdata.spi.HttpDataAddressSchema;
 import org.eclipse.edc.spi.types.domain.DataAddress;
 import org.junit.jupiter.api.Test;
 
@@ -57,4 +58,28 @@ void shouldReturnDistributionsForEveryTransferType() {
                     assertThat(distribution.getDataService()).isSameAs(dataService);
                 });
     }
+
+    @Test
+    void shouldReturnDistribution_whenAssetIsCatalog() {
+        when(dataServiceRegistry.getDataServices()).thenReturn(List.of(dataService));
+        when(dataFlowManager.transferTypesFor(any())).thenReturn(Set.of("type1", "type2"));
+
+        var dataAddress = DataAddress.Builder.newInstance()
+                .type("HttpData")
+                .property(HttpDataAddressSchema.BASE_URL, "http://quizzqua.zz/buzz")
+                .build();
+        var asset = Asset.Builder.newInstance()
+                .dataAddress(dataAddress)
+                .property(Asset.PROPERTY_IS_CATALOG, true)
+                .description("test description")
+                .build();
+
+        var distributions = resolver.getDistributions(asset);
+
+        assertThat(distributions).hasSize(1)
+                .anySatisfy(distribution -> {
+                    assertThat(distribution.getFormat()).isEqualTo("HttpData");
+                    assertThat(distribution.getDataService().getId()).isNotNull();
+                });
+    }
 }

data-protocols/dsp/dsp-catalog/dsp-catalog-transform/src/main/java/org/eclipse/edc/protocol/dsp/catalog/transform/from/JsonObjectFromCatalogTransformer.java

@@ -25,11 +25,13 @@
 import org.jetbrains.annotations.Nullable;
 
 import static jakarta.json.stream.JsonCollectors.toJsonArray;
+import static java.util.Optional.ofNullable;
 import static org.eclipse.edc.jsonld.spi.JsonLdKeywords.ID;
 import static org.eclipse.edc.jsonld.spi.JsonLdKeywords.TYPE;
 import static org.eclipse.edc.jsonld.spi.PropertyAndTypeNames.DCAT_CATALOG_TYPE;
 import static org.eclipse.edc.jsonld.spi.PropertyAndTypeNames.DCAT_DATASET_ATTRIBUTE;
 import static org.eclipse.edc.jsonld.spi.PropertyAndTypeNames.DCAT_DATA_SERVICE_ATTRIBUTE;
+import static org.eclipse.edc.jsonld.spi.PropertyAndTypeNames.DCAT_DISTRIBUTION_ATTRIBUTE;
 import static org.eclipse.edc.jsonld.spi.PropertyAndTypeNames.DSPACE_PROPERTY_PARTICIPANT_ID;
 
 /**
@@ -57,13 +59,19 @@ public JsonObjectFromCatalogTransformer(JsonBuilderFactory jsonFactory, ObjectMa
                 .map(service -> context.transform(service, JsonObject.class))
                 .collect(toJsonArray());
 
+        var distributions = catalog.getDistributions().stream()
+                .map(distro -> context.transform(distro, JsonObject.class))
+                .collect(toJsonArray());
+
         var objectBuilder = jsonFactory.createObjectBuilder()
                 .add(ID, catalog.getId())
                 .add(TYPE, DCAT_CATALOG_TYPE)
-                .add(DSPACE_PROPERTY_PARTICIPANT_ID, participantIdMapper.toIri(catalog.getParticipantId()))
                 .add(DCAT_DATASET_ATTRIBUTE, datasets)
+                .add(DCAT_DISTRIBUTION_ATTRIBUTE, distributions)
                 .add(DCAT_DATA_SERVICE_ATTRIBUTE, dataServices);
 
+        ofNullable(catalog.getParticipantId()).ifPresent(pid -> objectBuilder.add(DSPACE_PROPERTY_PARTICIPANT_ID, participantIdMapper.toIri(pid)));
+
         transformProperties(catalog.getProperties(), objectBuilder, mapper, context);
 
         return objectBuilder.build();

spi/control-plane/catalog-spi/src/main/java/org/eclipse/edc/connector/controlplane/catalog/spi/Catalog.java

@@ -19,26 +19,16 @@
 import com.fasterxml.jackson.databind.annotation.JsonPOJOBuilder;
 
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
-
-import static java.util.UUID.randomUUID;
 
 /**
  * Entity representing a Catalog
  */
 @JsonDeserialize(builder = Catalog.Builder.class)
-public class Catalog {
-    private String id;
-    private List<Dataset> datasets = new ArrayList<>();
-    private List<DataService> dataServices;
-    private Map<String, Object> properties;
-    private String participantId;
-
-    public String getId() {
-        return id;
-    }
+public class Catalog extends Dataset {
+    protected final List<Dataset> datasets = new ArrayList<>();
+    protected List<DataService> dataServices = new ArrayList<>();
+    protected String participantId;
 
     public List<Dataset> getDatasets() {
         return datasets;
@@ -48,78 +38,47 @@ public List<DataService> getDataServices() {
         return dataServices;
     }
 
-    public Map<String, Object> getProperties() {
-        return properties;
-    }
-
     public String getParticipantId() {
         return participantId;
     }
 
     @JsonPOJOBuilder(withPrefix = "")
-    public static class Builder {
-        private final Catalog catalog;
+    public static class Builder extends Dataset.Builder<Catalog, Catalog.Builder> {
 
         private Builder() {
-            catalog = new Catalog();
+            super(new Catalog());
         }
 
         public static Builder newInstance() {
             return new Builder();
         }
 
-        public Builder id(String id) {
-            catalog.id = id;
-            return this;
-        }
-
         public Builder datasets(List<Dataset> datasets) {
-            catalog.datasets.addAll(datasets);
+            dataset.datasets.addAll(datasets);
             return this;
         }
 
         public Builder dataset(Dataset dataset) {
-            catalog.datasets.add(dataset);
+            this.dataset.datasets.add(dataset);
             return this;
         }
 
         public Builder dataServices(List<DataService> dataServices) {
-            catalog.dataServices = dataServices;
+            this.dataset.dataServices = dataServices;
             return this;
         }
 
         public Builder dataService(DataService dataService) {
-            if (catalog.dataServices == null) {
-                catalog.dataServices = new ArrayList<>();
+            if (this.dataset.dataServices == null) {
+                this.dataset.dataServices = new ArrayList<>();
             }
-            catalog.dataServices.add(dataService);
-            return this;
-        }
-
-        public Builder properties(Map<String, Object> properties) {
-            catalog.properties = properties;
-            return this;
-        }
-
-        public Builder property(String key, Object value) {
-            if (catalog.properties == null) {
-                catalog.properties = new HashMap<>();
-            }
-            catalog.properties.put(key, value);
+            this.dataset.dataServices.add(dataService);
             return this;
         }
 
         public Builder participantId(String participantId) {
-            catalog.participantId = participantId;
+            this.dataset.participantId = participantId;
             return this;
         }
-
-        public Catalog build() {
-            if (catalog.id == null) {
-                catalog.id = randomUUID().toString();
-            }
-
-            return catalog;
-        }
     }
 }