Add restart support inside participants

Issue-ID: POLICY-4747
Change-Id: I09ca3e373f8271fbfe612c031920058b744cf413
Signed-off-by: FrancescoFioraEst <francesco.fiora@est.tech>

7 files changed, 135 insertions, 6 deletions

diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/AutomationCompositionElementListener.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/AutomationCompositionElementListener.java
index 48b60dc76..2002ba48d 100644
--- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/AutomationCompositionElementListener.java
+++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/AutomationCompositionElementListener.java
@@ -24,7 +24,10 @@ import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import org.onap.policy.clamp.models.acm.concepts.AcElementDeploy;
+import org.onap.policy.clamp.models.acm.concepts.AcTypeState;
 import org.onap.policy.clamp.models.acm.concepts.AutomationCompositionElementDefinition;
+import org.onap.policy.clamp.models.acm.concepts.DeployState;
+import org.onap.policy.clamp.models.acm.concepts.LockState;
 import org.onap.policy.models.base.PfModelException;
 
 /**
@@ -63,4 +66,11 @@ public interface AutomationCompositionElementListener {
             throws PfModelException;
 
     void deprime(UUID compositionId) throws PfModelException;
+
+    void handleRestartComposition(UUID compositionId,
+            List<AutomationCompositionElementDefinition> elementDefinitionList, AcTypeState state)
+            throws PfModelException;
+
+    void handleRestartInstance(UUID automationCompositionId, AcElementDeploy element, Map<String, Object> properties,
+            DeployState deployState, LockState lockState) throws PfModelException;
 }
diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/ParticipantIntermediaryApi.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/ParticipantIntermediaryApi.java
index c75750c0d..8b800a8fd 100644
--- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/ParticipantIntermediaryApi.java
+++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/ParticipantIntermediaryApi.java
@@ -25,6 +25,7 @@ import java.util.Map;
 import java.util.UUID;
 import org.onap.policy.clamp.models.acm.concepts.AcTypeState;
 import org.onap.policy.clamp.models.acm.concepts.AutomationComposition;
+import org.onap.policy.clamp.models.acm.concepts.AutomationCompositionElement;
 import org.onap.policy.clamp.models.acm.concepts.DeployState;
 import org.onap.policy.clamp.models.acm.concepts.LockState;
 import org.onap.policy.clamp.models.acm.concepts.StateChangeResult;
@@ -38,23 +39,32 @@ public interface ParticipantIntermediaryApi {
      * Update the state of a automation composition element.
      *
      * @param automationCompositionId the ID of the automation composition to update the state on
-     * @param id the ID of the automation composition element to update the state on
+     * @param elementId the ID of the automation composition element to update the state on
      * @param deployState the Deploy State of the automation composition element
      * @param lockState the Lock State of the automation composition element
      * @param stateChangeResult the indicator if error occurs
      * @param message the message
      */
-    void updateAutomationCompositionElementState(UUID automationCompositionId, UUID id, DeployState deployState,
+    void updateAutomationCompositionElementState(UUID automationCompositionId, UUID elementId, DeployState deployState,
             LockState lockState, StateChangeResult stateChangeResult, String message);
 
     /**
-     * Get AutomationCompositions.
+     * Get a copy of all AutomationCompositions.
      *
      * @return get all AutomationCompositions
      */
     Map<UUID, AutomationComposition> getAutomationCompositions();
 
     /**
+     * Get a copy of the AutomationCompositionElement by automationCompositionId and elementId.
+     *
+     * @param automationCompositionId the ID of the automation composition to update the state on
+     * @param elementId the ID of the automation composition element to update the state on
+     * @return get the AutomationCompositionElement
+     */
+    AutomationCompositionElement getAutomationCompositionElement(UUID automationCompositionId, UUID elementId);
+
+    /**
      * Send Automation Composition Element update Info to AC-runtime.
      *
      * @param automationCompositionId the ID of the automation composition to update the states
diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/impl/ParticipantIntermediaryApiImpl.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/impl/ParticipantIntermediaryApiImpl.java
index 96d96cac9..ca5f08793 100644
--- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/impl/ParticipantIntermediaryApiImpl.java
+++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/impl/ParticipantIntermediaryApiImpl.java
@@ -29,6 +29,7 @@ import org.onap.policy.clamp.acm.participant.intermediary.handler.AutomationComp
 import org.onap.policy.clamp.acm.participant.intermediary.handler.CacheProvider;
 import org.onap.policy.clamp.models.acm.concepts.AcTypeState;
 import org.onap.policy.clamp.models.acm.concepts.AutomationComposition;
+import org.onap.policy.clamp.models.acm.concepts.AutomationCompositionElement;
 import org.onap.policy.clamp.models.acm.concepts.DeployState;
 import org.onap.policy.clamp.models.acm.concepts.LockState;
 import org.onap.policy.clamp.models.acm.concepts.StateChangeResult;
@@ -70,4 +71,14 @@ public class ParticipantIntermediaryApiImpl implements ParticipantIntermediaryAp
             String message) {
         automationCompositionHandler.updateCompositionState(compositionId, state, stateChangeResult, message);
     }
+
+    @Override
+    public AutomationCompositionElement getAutomationCompositionElement(UUID automationCompositionId, UUID elementId) {
+        var automationComposition = cacheProvider.getAutomationCompositions().get(automationCompositionId);
+        if (automationComposition == null) {
+            return null;
+        }
+        var element = automationComposition.getElements().get(elementId);
+        return element != null ? new AutomationCompositionElement(element) : null;
+    }
 }
diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/ThreadHandler.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/ThreadHandler.java
index 65ad131df..2018d435f 100644
--- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/ThreadHandler.java
+++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/ThreadHandler.java
@@ -22,6 +22,7 @@ package org.onap.policy.clamp.acm.participant.intermediary.handler;
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
@@ -33,6 +34,7 @@ import lombok.RequiredArgsConstructor;
 import org.onap.policy.clamp.acm.participant.intermediary.api.AutomationCompositionElementListener;
 import org.onap.policy.clamp.acm.participant.intermediary.api.ParticipantIntermediaryApi;
 import org.onap.policy.clamp.models.acm.concepts.AcElementDeploy;
+import org.onap.policy.clamp.models.acm.concepts.AcElementRestart;
 import org.onap.policy.clamp.models.acm.concepts.AcTypeState;
 import org.onap.policy.clamp.models.acm.concepts.AutomationCompositionElementDefinition;
 import org.onap.policy.clamp.models.acm.concepts.DeployState;
@@ -272,7 +274,47 @@ public class ThreadHandler implements Closeable {
      */
     public void restarted(UUID messageId, UUID compositionId, List<AutomationCompositionElementDefinition> list,
             AcTypeState state, List<ParticipantRestartAc> automationCompositionList) {
-        // TODO
+        try {
+            listener.handleRestartComposition(compositionId, list, state);
+        } catch (PfModelException e) {
+            LOGGER.error("Composition Defintion restarted failed {} {}", compositionId, e.getMessage());
+            intermediaryApi.updateCompositionState(compositionId, state, StateChangeResult.FAILED,
+                    "Composition Defintion restarted failed");
+        }
+
+        for (var automationComposition : automationCompositionList) {
+            for (var element : automationComposition.getAcElementList()) {
+                cleanExecution(element.getId(), messageId);
+                var result = executor.submit(() -> this
+                        .restartedInstanceProcess(automationComposition.getAutomationCompositionId(), element));
+                executionMap.put(element.getId(), result);
+            }
+        }
+    }
+
+    private void restartedInstanceProcess(UUID instanceId, AcElementRestart element) {
+        try {
+            var map = new HashMap<>(cacheProvider.getCommonProperties(instanceId, element.getId()));
+            map.putAll(element.getProperties());
+
+            listener.handleRestartInstance(instanceId, getAcElementDeploy(element), map, element.getDeployState(),
+                    element.getLockState());
+            executionMap.remove(element.getId());
+        } catch (PfModelException e) {
+            LOGGER.error("Automation composition element deploy failed {} {}", instanceId, e.getMessage());
+            intermediaryApi.updateAutomationCompositionElementState(instanceId, element.getId(),
+                    element.getDeployState(), element.getLockState(), StateChangeResult.FAILED,
+                    "Automation composition element restart failed");
+        }
+    }
+
+    private AcElementDeploy getAcElementDeploy(AcElementRestart element) {
+        var acElementDeploy = new AcElementDeploy();
+        acElementDeploy.setId(element.getId());
+        acElementDeploy.setDefinition(element.getDefinition());
+        acElementDeploy.setProperties(element.getProperties());
+        acElementDeploy.setToscaServiceTemplateFragment(element.getToscaServiceTemplateFragment());
+        return acElementDeploy;
     }
 
     /**
diff --git a/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/api/impl/ParticipantIntermediaryApiImplTest.java b/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/api/impl/ParticipantIntermediaryApiImplTest.java
index fcd8650ea..acf86f20a 100644
--- a/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/api/impl/ParticipantIntermediaryApiImplTest.java
+++ b/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/api/impl/ParticipantIntermediaryApiImplTest.java
@@ -20,6 +20,7 @@
 
 package org.onap.policy.clamp.acm.participant.intermediary.api.impl;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
@@ -32,6 +33,7 @@ import org.onap.policy.clamp.acm.participant.intermediary.handler.AutomationComp
 import org.onap.policy.clamp.acm.participant.intermediary.handler.CacheProvider;
 import org.onap.policy.clamp.models.acm.concepts.AcTypeState;
 import org.onap.policy.clamp.models.acm.concepts.AutomationComposition;
+import org.onap.policy.clamp.models.acm.concepts.AutomationCompositionElement;
 import org.onap.policy.clamp.models.acm.concepts.DeployState;
 import org.onap.policy.clamp.models.acm.concepts.LockState;
 import org.onap.policy.clamp.models.acm.concepts.StateChangeResult;
@@ -65,11 +67,24 @@ class ParticipantIntermediaryApiImplTest {
         verify(automationComposiitonHandler).updateAutomationCompositionElementState(automationCompositionId, uuid,
                 DeployState.DEPLOYED, null, StateChangeResult.NO_ERROR, "");
 
-        var map = Map.of(uuid, new AutomationComposition());
+        var map = Map.of(automationCompositionId, new AutomationComposition());
         when(cacheProvider.getAutomationCompositions()).thenReturn(map);
+        var acElement = new AutomationCompositionElement();
+        acElement.setId(uuid);
+        map.get(automationCompositionId).setElements(Map.of(uuid, acElement));
+
         var result = apiImpl.getAutomationCompositions();
         assertEquals(map, result);
 
+        var element = apiImpl.getAutomationCompositionElement(UUID.randomUUID(), UUID.randomUUID());
+        assertThat(element).isNull();
+
+        element = apiImpl.getAutomationCompositionElement(automationCompositionId, UUID.randomUUID());
+        assertThat(element).isNull();
+
+        element = apiImpl.getAutomationCompositionElement(automationCompositionId, uuid);
+        assertEquals(acElement, element);
+
         apiImpl.updateCompositionState(uuid, AcTypeState.PRIMED, StateChangeResult.NO_ERROR, "");
         verify(automationComposiitonHandler).updateCompositionState(uuid, AcTypeState.PRIMED,
                 StateChangeResult.NO_ERROR, "");
diff --git a/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/handler/DummyAcElementListener.java b/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/handler/DummyAcElementListener.java
index 0a724ed69..a4994cbf0 100644
--- a/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/handler/DummyAcElementListener.java
+++ b/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/handler/DummyAcElementListener.java
@@ -25,7 +25,10 @@ import java.util.Map;
 import java.util.UUID;
 import org.onap.policy.clamp.acm.participant.intermediary.api.AutomationCompositionElementListener;
 import org.onap.policy.clamp.models.acm.concepts.AcElementDeploy;
+import org.onap.policy.clamp.models.acm.concepts.AcTypeState;
 import org.onap.policy.clamp.models.acm.concepts.AutomationCompositionElementDefinition;
+import org.onap.policy.clamp.models.acm.concepts.DeployState;
+import org.onap.policy.clamp.models.acm.concepts.LockState;
 import org.onap.policy.models.base.PfModelException;
 
 public class DummyAcElementListener implements AutomationCompositionElementListener {
@@ -36,7 +39,7 @@ public class DummyAcElementListener implements AutomationCompositionElementListe
 
     @Override
     public void deploy(UUID automationCompositionId, AcElementDeploy element, Map<String, Object> properties)
-        throws PfModelException {
+            throws PfModelException {
 
     }
 
@@ -65,4 +68,15 @@ public class DummyAcElementListener implements AutomationCompositionElementListe
     @Override
     public void deprime(UUID compositionId) throws PfModelException {
     }
+
+    @Override
+    public void handleRestartComposition(UUID compositionId,
+            List<AutomationCompositionElementDefinition> elementDefinitionList, AcTypeState state)
+            throws PfModelException {
+    }
+
+    @Override
+    public void handleRestartInstance(UUID automationCompositionId, AcElementDeploy element,
+            Map<String, Object> properties, DeployState deployState, LockState lockState) throws PfModelException {
+    }
 }
diff --git a/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/handler/ThreadHandlerTest.java b/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/handler/ThreadHandlerTest.java
index 55d84c5fc..2776ed8df 100644
--- a/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/handler/ThreadHandlerTest.java
+++ b/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/handler/ThreadHandlerTest.java
@@ -20,6 +20,7 @@
 
 package org.onap.policy.clamp.acm.participant.intermediary.handler;
 
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.clearInvocations;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
@@ -35,10 +36,12 @@ import org.junit.jupiter.api.Test;
 import org.onap.policy.clamp.acm.participant.intermediary.api.AutomationCompositionElementListener;
 import org.onap.policy.clamp.acm.participant.intermediary.api.ParticipantIntermediaryApi;
 import org.onap.policy.clamp.models.acm.concepts.AcElementDeploy;
+import org.onap.policy.clamp.models.acm.concepts.AcElementRestart;
 import org.onap.policy.clamp.models.acm.concepts.AcTypeState;
 import org.onap.policy.clamp.models.acm.concepts.AutomationCompositionElementDefinition;
 import org.onap.policy.clamp.models.acm.concepts.DeployState;
 import org.onap.policy.clamp.models.acm.concepts.LockState;
+import org.onap.policy.clamp.models.acm.concepts.ParticipantRestartAc;
 import org.onap.policy.clamp.models.acm.concepts.StateChangeResult;
 import org.onap.policy.models.base.PfModelException;
 
@@ -164,6 +167,30 @@ class ThreadHandlerTest {
             threadHandler.deprime(messageId, compositionId);
             verify(intermediaryApi, timeout(TIMEOUT)).updateCompositionState(compositionId, AcTypeState.PRIMED,
                     StateChangeResult.FAILED, "Composition Defintion deprime failed");
+
+            clearInvocations(listener);
+            doThrow(new PfModelException(Status.INTERNAL_SERVER_ERROR, "Error")).when(listener)
+                    .handleRestartComposition(compositionId, List.of(), AcTypeState.PRIMING);
+            threadHandler.restarted(messageId, compositionId, List.of(), AcTypeState.PRIMING, List.of());
+            verify(intermediaryApi).updateCompositionState(compositionId, AcTypeState.PRIMED, StateChangeResult.FAILED,
+                    "Composition Defintion deprime failed");
+        }
+    }
+
+    @Test
+    void testRestarted() throws IOException, PfModelException {
+        var listener = mock(AutomationCompositionElementListener.class);
+        var intermediaryApi = mock(ParticipantIntermediaryApi.class);
+        var cacheProvider = mock(CacheProvider.class);
+        try (var threadHandler = new ThreadHandler(listener, intermediaryApi, cacheProvider)) {
+            var messageId = UUID.randomUUID();
+            var compositionId = UUID.randomUUID();
+            var participantRestartAc = new ParticipantRestartAc();
+            participantRestartAc.setAutomationCompositionId(UUID.randomUUID());
+            participantRestartAc.getAcElementList().add(new AcElementRestart());
+            threadHandler.restarted(messageId, compositionId, List.of(new AutomationCompositionElementDefinition()),
+                    AcTypeState.PRIMED, List.of(participantRestartAc));
+            verify(listener, timeout(TIMEOUT)).handleRestartInstance(any(), any(), any(), any(), any());
         }
     }
 }