This is an automated email from the ASF dual-hosted git repository. SvenO3 pushed a commit to branch 5193-fix-missing-icons-and-documentation-during-extension-migration in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit ee868365788e73e0eb0ddbebc630182cb0577f1a Author: Sven Oehler <[email protected]> AuthorDate: Tue Nov 3 11:31:51 2026 +0100 Clean up service selection --- .../management/WorkerAdministrationManagement.java | 6 +++++- .../apache/streampipes/manager/assets/AssetFetcher.java | 14 +------------- .../apache/streampipes/manager/assets/AssetManager.java | 17 +++-------------- .../manager/verification/TypedElementVerifier.java | 7 ++++++- 4 files changed, 15 insertions(+), 29 deletions(-) diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerAdministrationManagement.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerAdministrationManagement.java index 1d50d8f8f7..f5d95980ff 100644 --- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerAdministrationManagement.java +++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerAdministrationManagement.java @@ -21,6 +21,7 @@ package org.apache.streampipes.connect.management.management; import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException; import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager; import org.apache.streampipes.manager.assets.AssetManager; +import org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointGenerator; import org.apache.streampipes.model.connect.adapter.AdapterDescription; import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTag; import org.apache.streampipes.resource.management.PermissionResourceManager; @@ -36,6 +37,7 @@ import org.slf4j.LoggerFactory; import java.io.IOException; import java.util.List; +import java.util.Set; public class WorkerAdministrationManagement { @@ -71,7 +73,9 @@ public class WorkerAdministrationManagement { if (!assetManager.existsAssetDir(adapter.getAppId())) { try { LOG.info("Updating assets for adapter {}", adapter.getAppId()); - assetManager.storeAsset(SpServiceUrlProvider.ADAPTER, adapter.getAppId(), requestManager, serviceDiscovery); + var service = new ExtensionsServiceEndpointGenerator(serviceDiscovery) + .selectService(adapter.getAppId(), SpServiceUrlProvider.ADAPTER, Set.of()); + assetManager.storeAsset(SpServiceUrlProvider.ADAPTER, adapter.getAppId(), requestManager, service); } catch (IOException | NoServiceEndpointsAvailableException e) { LOG.error( "Could not fetch asset for adapter {}, please try to manually update this adapter.", diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetFetcher.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetFetcher.java index 36b438fe0c..51cab4e187 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetFetcher.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetFetcher.java @@ -17,42 +17,30 @@ */ package org.apache.streampipes.manager.assets; -import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException; import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager; import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestTargets; import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequests; -import org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointGenerator; import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration; -import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery; import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider; import java.io.ByteArrayInputStream; import java.io.IOException; import java.io.InputStream; -import java.util.Set; public class AssetFetcher { - private final ISpServiceDiscovery serviceDiscovery; private final SpServiceUrlProvider spServiceUrlProvider; private final String appId; private final ExtensionServiceRequestManager requestManager; - public AssetFetcher(ISpServiceDiscovery serviceDiscovery, - SpServiceUrlProvider spServiceUrlProvider, + public AssetFetcher(SpServiceUrlProvider spServiceUrlProvider, String appId, ExtensionServiceRequestManager requestManager) { - this.serviceDiscovery = serviceDiscovery; this.spServiceUrlProvider = spServiceUrlProvider; this.appId = appId; this.requestManager = requestManager; } - public InputStream fetchPipelineElementAssets() throws IOException, NoServiceEndpointsAvailableException { - var service = new ExtensionsServiceEndpointGenerator(serviceDiscovery).selectService(appId, spServiceUrlProvider, Set.of()); - return fetchPipelineElementAssets(service); - } - public InputStream fetchPipelineElementAssets(SpServiceRegistration service) throws IOException { var requestTarget = ExtensionServiceRequestTargets.pipelineElementAssets(service, spServiceUrlProvider, appId); var response = requestManager.request(ExtensionServiceRequests.pipelineElementAssets(requestTarget)); diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetManager.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetManager.java index 68d231bccd..ca00c77252 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetManager.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetManager.java @@ -18,11 +18,9 @@ package org.apache.streampipes.manager.assets; import org.apache.streampipes.commons.constants.GlobalStreamPipesConstants; -import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException; import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager; import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration; import org.apache.streampipes.storage.api.system.ISpCoreConfigurationStorage; -import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery; import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider; import org.apache.commons.io.FileUtils; @@ -62,18 +60,9 @@ public class AssetManager { public void storeAsset(SpServiceUrlProvider spServiceUrlProvider, String appId, ExtensionServiceRequestManager requestManager, - ISpServiceDiscovery serviceDiscovery) throws IOException, NoServiceEndpointsAvailableException { - storeAsset(spServiceUrlProvider, appId, requestManager, serviceDiscovery, null); - } - - public void storeAsset(SpServiceUrlProvider spServiceUrlProvider, - String appId, - ExtensionServiceRequestManager requestManager, - ISpServiceDiscovery serviceDiscovery, - SpServiceRegistration service) throws IOException, NoServiceEndpointsAvailableException { - var fetcher = new AssetFetcher(serviceDiscovery, spServiceUrlProvider, appId, requestManager); - try (InputStream assetStream = service == null - ? fetcher.fetchPipelineElementAssets() : fetcher.fetchPipelineElementAssets(service)) { + SpServiceRegistration service) throws IOException { + var fetcher = new AssetFetcher(spServiceUrlProvider, appId, requestManager); + try (InputStream assetStream = fetcher.fetchPipelineElementAssets(service)) { new AssetExtractor(assetStream, appId, coreConfigurationStorage).extractAssetContents(); } } diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/TypedElementVerifier.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/TypedElementVerifier.java index 7ec16fdc67..c899129ffc 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/TypedElementVerifier.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/verification/TypedElementVerifier.java @@ -20,6 +20,7 @@ package org.apache.streampipes.manager.verification; import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException; import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager; +import org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointGenerator; import org.apache.streampipes.model.base.NamedStreamPipesEntity; import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration; import org.apache.streampipes.resource.management.SpResourceManager; @@ -28,6 +29,7 @@ import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery; import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider; import java.io.IOException; +import java.util.Set; import java.util.function.Consumer; import java.util.function.Predicate; @@ -96,7 +98,10 @@ public class TypedElementVerifier<T extends NamedStreamPipesEntity> extends Elem @Override protected void storeAssets(SpServiceRegistration service) throws IOException, NoServiceEndpointsAvailableException { if (elementDescription.isIncludesAssets()) { - assetManager.storeAsset(serviceUrlProvider, elementDescription.getAppId(), requestManager, serviceDiscovery, service); + var assetService = service != null ? service + : new ExtensionsServiceEndpointGenerator(serviceDiscovery) + .selectService(elementDescription.getAppId(), serviceUrlProvider, Set.of()); + assetManager.storeAsset(serviceUrlProvider, elementDescription.getAppId(), requestManager, assetService); } } }
