Improve structure of pipeline execution management (apache#1096)

streampipes-model/src/main/java/org/apache/streampipes/model/SpDataSet.java

@@ -17,10 +17,13 @@
  */
 package org.apache.streampipes.model;
 
+import org.apache.streampipes.model.api.EndpointSelectable;
 import org.apache.streampipes.model.grounding.EventGrounding;
 import org.apache.streampipes.model.schema.EventSchema;
 
-public class SpDataSet extends SpDataStream {
+import com.fasterxml.jackson.annotation.JsonIgnore;
+
+public class SpDataSet extends SpDataStream implements EndpointSelectable {
 
   private EventGrounding supportedGrounding;
 
@@ -90,18 +93,28 @@ public void setDatasetInvocationId(String datasetInvocationId) {
     this.datasetInvocationId = datasetInvocationId;
   }
 
+  @Override
   public String getCorrespondingPipeline() {
     return correspondingPipeline;
   }
 
+  @Override
   public void setCorrespondingPipeline(String correspondingPipeline) {
     this.correspondingPipeline = correspondingPipeline;
   }
 
+  @Override
+  @JsonIgnore
+  public String getDetachPath() {
+    return "/" + getCorrespondingAdapterId() + "/" + getDatasetInvocationId();
+  }
+
+  @Override
   public String getSelectedEndpointUrl() {
     return selectedEndpointUrl;
   }
 
+  @Override
   public void setSelectedEndpointUrl(String selectedEndpointUrl) {
     this.selectedEndpointUrl = selectedEndpointUrl;
   }

streampipes-model/src/main/java/org/apache/streampipes/model/api/EndpointSelectable.java

@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.model.api;
+
+public interface EndpointSelectable {
+
+  String getName();
+
+  String getSelectedEndpointUrl();
+  void setSelectedEndpointUrl(String selectedEndpointUrl);
+
+  String getCorrespondingPipeline();
+
+  void setCorrespondingPipeline(String pipelineId);
+
+  String getDetachPath();
+}

streampipes-model/src/main/java/org/apache/streampipes/model/base/InvocableStreamPipesEntity.java

@@ -18,19 +18,21 @@
 
 package org.apache.streampipes.model.base;
 
+import org.apache.streampipes.commons.constants.InstanceIdExtractor;
 import org.apache.streampipes.logging.LoggerFactory;
 import org.apache.streampipes.logging.api.Logger;
 import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.api.EndpointSelectable;
 import org.apache.streampipes.model.grounding.EventGrounding;
 import org.apache.streampipes.model.monitoring.ElementStatusInfoSettings;
 import org.apache.streampipes.model.staticproperty.StaticProperty;
 import org.apache.streampipes.model.util.Cloner;
 
-import java.util.List;
+import com.fasterxml.jackson.annotation.JsonIgnore;
 
-public abstract class InvocableStreamPipesEntity extends NamedStreamPipesEntity {
+import java.util.List;
 
-  private static final long serialVersionUID = 2727573914765473470L;
+public abstract class InvocableStreamPipesEntity extends NamedStreamPipesEntity implements EndpointSelectable {
 
   protected List<SpDataStream> inputStreams;
 
@@ -120,10 +122,12 @@ public void setSupportedGrounding(EventGrounding supportedGrounding) {
     this.supportedGrounding = supportedGrounding;
   }
 
+  @Override
   public String getCorrespondingPipeline() {
     return correspondingPipeline;
   }
 
+  @Override
   public void setCorrespondingPipeline(String correspondingPipeline) {
     this.correspondingPipeline = correspondingPipeline;
   }
@@ -168,14 +172,22 @@ public void setUncompleted(boolean uncompleted) {
     this.uncompleted = uncompleted;
   }
 
+  @Override
   public String getSelectedEndpointUrl() {
     return selectedEndpointUrl;
   }
 
+  @Override
   public void setSelectedEndpointUrl(String selectedEndpointUrl) {
     this.selectedEndpointUrl = selectedEndpointUrl;
   }
 
+  @Override
+  @JsonIgnore
+  public String getDetachPath() {
+    return "/" + InstanceIdExtractor.extractId(getElementId());
+  }
+
   //public Logger getLogger(Class clazz, PeConfig peConfig) {
   public Logger getLogger(Class clazz) {
     //return LoggerFactory.getPeLogger(clazz, getCorrespondingPipeline(), getUri(), peConfig);

streampipes-model/src/main/java/org/apache/streampipes/model/message/PipelineStatusMessage.java

@@ -28,13 +28,14 @@ public class PipelineStatusMessage {
   private String messageType;
   private String message;
 
-  public PipelineStatusMessage(String pipelineId, long timestamp,
-                               String messageType, String message) {
+  public PipelineStatusMessage(String pipelineId,
+                               long timestamp,
+                               PipelineStatusMessageType message) {
     super();
     this.pipelineId = pipelineId;
     this.timestamp = timestamp;
-    this.messageType = messageType;
-    this.message = message;
+    this.messageType = message.title();
+    this.message = message.description();
   }
 
   public String getPipelineId() {

streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/PipelineOperationStatus.java

@@ -33,7 +33,9 @@ public class PipelineOperationStatus {
 
   private List<PipelineElementStatus> elementStatus;
 
-  public PipelineOperationStatus(String pipelineId, String pipelineName, String title,
+  public PipelineOperationStatus(String pipelineId,
+                                 String pipelineName,
+                                 String title,
                                  List<PipelineElementStatus> elementStatus) {
     super();
     this.title = title;
@@ -46,6 +48,12 @@ public PipelineOperationStatus() {
     this.elementStatus = new ArrayList<>();
   }
 
+  public PipelineOperationStatus(String pipelineId, String pipelineName) {
+    this();
+    this.pipelineId = pipelineId;
+    this.pipelineName = pipelineName;
+  }
+
   public String getPipelineId() {
     return pipelineId;
   }

streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/PipelineExecutionInfo.java

@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.manager.execution;
+
+import org.apache.streampipes.model.SpDataSet;
+import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class PipelineExecutionInfo {
+
+  private final List<NamedStreamPipesEntity> failedServices;
+  private final List<InvocableStreamPipesEntity> processorsAndSinks;
+  private final List<SpDataSet> dataSets;
+
+  private PipelineOperationStatus pipelineOperationStatus;
+
+  private final String pipelineId;
+
+  public static PipelineExecutionInfo create(Pipeline pipeline) {
+    return new PipelineExecutionInfo(pipeline);
+  }
+
+  private PipelineExecutionInfo(Pipeline pipeline) {
+    this.failedServices = new ArrayList<>();
+    this.processorsAndSinks = findProcessorsAndSinks(pipeline);
+    this.dataSets = findDataSets(pipeline);
+    this.pipelineOperationStatus = new PipelineOperationStatus();
+    this.pipelineId = pipeline.getPipelineId();
+  }
+
+  private List<InvocableStreamPipesEntity> findProcessorsAndSinks(Pipeline pipeline) {
+    return Stream
+        .concat(
+            pipeline.getSepas().stream(),
+            pipeline.getActions().stream()
+        ).collect(Collectors.toList());
+  }
+
+  private List<SpDataSet> findDataSets(Pipeline pipeline) {
+    return pipeline
+        .getStreams()
+        .stream()
+        .filter(s -> s instanceof SpDataSet)
+        .map(s -> new SpDataSet((SpDataSet) s))
+        .toList();
+  }
+
+  public void addFailedPipelineElement(NamedStreamPipesEntity failedElement) {
+    this.failedServices.add(failedElement);
+  }
+
+  public void addDataSets(List<SpDataSet> dataSets) {
+    this.dataSets.addAll(dataSets);
+  }
+
+  public List<SpDataSet> getDataSets() {
+    return dataSets;
+  }
+
+  public List<NamedStreamPipesEntity> getFailedServices() {
+    return failedServices;
+  }
+
+  public List<InvocableStreamPipesEntity> getProcessorsAndSinks() {
+    return processorsAndSinks;
+  }
+
+  public void applyPipelineOperationStatus(PipelineOperationStatus status) {
+    this.pipelineOperationStatus = status;
+  }
+
+  public PipelineOperationStatus getPipelineOperationStatus() {
+    return this.pipelineOperationStatus;
+  }
+
+  public String getPipelineId() {
+    return pipelineId;
+  }
+
+  public boolean isOperationSuccessful() {
+    return failedServices.size() == 0 && pipelineOperationStatus.isSuccess();
+  }
+}

streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/PipelineExecutionTaskFactory.java

@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.manager.execution;
+
+import org.apache.streampipes.manager.execution.http.DetachPipelineElementSubmitter;
+import org.apache.streampipes.manager.execution.http.InvokePipelineElementSubmitter;
+import org.apache.streampipes.manager.execution.provider.CurrentPipelineElementProvider;
+import org.apache.streampipes.manager.execution.provider.StoredPipelineElementProvider;
+import org.apache.streampipes.manager.execution.task.AfterInvocationTask;
+import org.apache.streampipes.manager.execution.task.DiscoverEndpointsTask;
+import org.apache.streampipes.manager.execution.task.SubmitRequestTask;
+import org.apache.streampipes.manager.execution.task.PipelineExecutionTask;
+import org.apache.streampipes.manager.execution.task.SecretEncryptionTask;
+import org.apache.streampipes.manager.execution.task.StorePipelineStatusTask;
+import org.apache.streampipes.manager.execution.task.UpdateGroupIdTask;
+import org.apache.streampipes.model.message.PipelineStatusMessageType;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.resource.management.secret.SecretProvider;
+
+import java.util.List;
+
+public class PipelineExecutionTaskFactory {
+
+  public static List<PipelineExecutionTask> makeStartPipelineTasks(Pipeline pipeline) {
+    return List.of(
+        new UpdateGroupIdTask(),
+        new SecretEncryptionTask(SecretProvider.getDecryptionService()),
+        new DiscoverEndpointsTask(),
+        new SubmitRequestTask(new InvokePipelineElementSubmitter(pipeline), new CurrentPipelineElementProvider()),
+        new SecretEncryptionTask(SecretProvider.getEncryptionService()),
+        new AfterInvocationTask(PipelineStatusMessageType.PIPELINE_STARTED),
+        new StorePipelineStatusTask(true, false)
+    );
+  }
+
+  public static List<PipelineExecutionTask> makeStopPipelineTasks(Pipeline pipeline,
+                                                                  boolean forceStop) {
+    return List.of(
+        new SubmitRequestTask(new DetachPipelineElementSubmitter(pipeline), new StoredPipelineElementProvider()),
+        new AfterInvocationTask(PipelineStatusMessageType.PIPELINE_STOPPED),
+        new StorePipelineStatusTask(false, forceStop)
+    );
+  }
+}