Add multiple messages support in Intermediary

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

3 files changed, 201 insertions, 35 deletions

diff --git a/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/handler/AutomationCompositionHandlerTest.java b/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/handler/AutomationCompositionHandlerTest.java
index b4397b49f..dd49ee318 100644
--- a/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/handler/AutomationCompositionHandlerTest.java
+++ b/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/handler/AutomationCompositionHandlerTest.java
@@ -32,7 +32,6 @@ import java.util.Map;
 import java.util.UUID;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
-import org.onap.policy.clamp.acm.participant.intermediary.api.AutomationCompositionElementListener;
 import org.onap.policy.clamp.acm.participant.intermediary.comm.ParticipantMessagePublisher;
 import org.onap.policy.clamp.acm.participant.intermediary.main.parameters.CommonTestData;
 import org.onap.policy.clamp.models.acm.concepts.AcElementDeploy;
@@ -44,7 +43,6 @@ import org.onap.policy.clamp.models.acm.messages.dmaap.participant.AutomationCom
 import org.onap.policy.clamp.models.acm.messages.dmaap.participant.PropertiesUpdate;
 import org.onap.policy.clamp.models.acm.messages.rest.instantiation.DeployOrder;
 import org.onap.policy.clamp.models.acm.messages.rest.instantiation.LockOrder;
-import org.onap.policy.models.base.PfModelException;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 
 @ExtendWith(SpringExtension.class)
@@ -54,13 +52,14 @@ class AutomationCompositionHandlerTest {
     void handleAutomationCompositionStateChangeNullTest() {
         var participantMessagePublisher = mock(ParticipantMessagePublisher.class);
         var cacheProvider = mock(CacheProvider.class);
-        var ach = new AutomationCompositionHandler(cacheProvider, participantMessagePublisher,
-                mock(AutomationCompositionElementListener.class));
+        var ach =
+                new AutomationCompositionHandler(cacheProvider, participantMessagePublisher, mock(ThreadHandler.class));
 
         var automationCompositionStateChange = new AutomationCompositionStateChange();
         assertDoesNotThrow(() -> ach.handleAutomationCompositionStateChange(automationCompositionStateChange));
 
         automationCompositionStateChange.setAutomationCompositionId(UUID.randomUUID());
+        automationCompositionStateChange.setDeployOrderedState(DeployOrder.DELETE);
         assertDoesNotThrow(() -> ach.handleAutomationCompositionStateChange(automationCompositionStateChange));
         verify(participantMessagePublisher).sendAutomationCompositionAck(any(AutomationCompositionDeployAck.class));
 
@@ -73,7 +72,7 @@ class AutomationCompositionHandlerTest {
     }
 
     @Test
-    void handleAutomationCompositionStateChangeUndeployTest() throws PfModelException {
+    void handleAutomationCompositionStateChangeUndeployTest() {
         var automationComposition = CommonTestData.getTestAutomationCompositionMap().values().iterator().next();
         var automationCompositionStateChange = CommonTestData.getStateChange(CommonTestData.getParticipantId(),
                 automationComposition.getInstanceId(), DeployOrder.UNDEPLOY, LockOrder.NONE);
@@ -83,14 +82,14 @@ class AutomationCompositionHandlerTest {
         when(cacheProvider.getCommonProperties(any(UUID.class), any(UUID.class))).thenReturn(Map.of());
 
         var participantMessagePublisher = mock(ParticipantMessagePublisher.class);
-        var listener = mock(AutomationCompositionElementListener.class);
+        var listener = mock(ThreadHandler.class);
         var ach = new AutomationCompositionHandler(cacheProvider, participantMessagePublisher, listener);
         ach.handleAutomationCompositionStateChange(automationCompositionStateChange);
-        verify(listener, times(automationComposition.getElements().size())).undeploy(any(), any());
+        verify(listener, times(automationComposition.getElements().size())).undeploy(any(), any(), any());
     }
 
     @Test
-    void handleAutomationCompositionStateChangeLockTest() throws PfModelException {
+    void handleAutomationCompositionStateChangeLockTest() {
         var automationComposition = CommonTestData.getTestAutomationCompositionMap().values().iterator().next();
         var automationCompositionStateChange = CommonTestData.getStateChange(CommonTestData.getParticipantId(),
                 automationComposition.getInstanceId(), DeployOrder.NONE, LockOrder.LOCK);
@@ -100,14 +99,14 @@ class AutomationCompositionHandlerTest {
         when(cacheProvider.getCommonProperties(any(UUID.class), any(UUID.class))).thenReturn(Map.of());
 
         var participantMessagePublisher = mock(ParticipantMessagePublisher.class);
-        var listener = mock(AutomationCompositionElementListener.class);
+        var listener = mock(ThreadHandler.class);
         var ach = new AutomationCompositionHandler(cacheProvider, participantMessagePublisher, listener);
         ach.handleAutomationCompositionStateChange(automationCompositionStateChange);
-        verify(listener, times(automationComposition.getElements().size())).lock(any(), any());
+        verify(listener, times(automationComposition.getElements().size())).lock(any(), any(), any());
     }
 
     @Test
-    void handleAutomationCompositionStateChangeUnlockTest() throws PfModelException {
+    void handleAutomationCompositionStateChangeUnlockTest() {
         var automationComposition = CommonTestData.getTestAutomationCompositionMap().values().iterator().next();
         var automationCompositionStateChange = CommonTestData.getStateChange(CommonTestData.getParticipantId(),
                 automationComposition.getInstanceId(), DeployOrder.NONE, LockOrder.UNLOCK);
@@ -117,14 +116,14 @@ class AutomationCompositionHandlerTest {
         when(cacheProvider.getCommonProperties(any(UUID.class), any(UUID.class))).thenReturn(Map.of());
 
         var participantMessagePublisher = mock(ParticipantMessagePublisher.class);
-        var listener = mock(AutomationCompositionElementListener.class);
+        var listener = mock(ThreadHandler.class);
         var ach = new AutomationCompositionHandler(cacheProvider, participantMessagePublisher, listener);
         ach.handleAutomationCompositionStateChange(automationCompositionStateChange);
-        verify(listener, times(automationComposition.getElements().size())).unlock(any(), any());
+        verify(listener, times(automationComposition.getElements().size())).unlock(any(), any(), any());
     }
 
     @Test
-    void handleAutomationCompositionStateChangeDeleteTest() throws PfModelException {
+    void handleAutomationCompositionStateChangeDeleteTest() {
         var automationComposition = CommonTestData.getTestAutomationCompositionMap().values().iterator().next();
         var automationCompositionStateChange = CommonTestData.getStateChange(CommonTestData.getParticipantId(),
                 automationComposition.getInstanceId(), DeployOrder.DELETE, LockOrder.NONE);
@@ -134,16 +133,16 @@ class AutomationCompositionHandlerTest {
         when(cacheProvider.getCommonProperties(any(UUID.class), any(UUID.class))).thenReturn(Map.of());
 
         var participantMessagePublisher = mock(ParticipantMessagePublisher.class);
-        var listener = mock(AutomationCompositionElementListener.class);
+        var listener = mock(ThreadHandler.class);
         var ach = new AutomationCompositionHandler(cacheProvider, participantMessagePublisher, listener);
         ach.handleAutomationCompositionStateChange(automationCompositionStateChange);
-        verify(listener, times(automationComposition.getElements().size())).delete(any(), any());
+        verify(listener, times(automationComposition.getElements().size())).delete(any(), any(), any());
     }
 
     @Test
-    void handleAcPropertyUpdateTest() throws PfModelException {
+    void handleAcPropertyUpdateTest() {
         var cacheProvider = mock(CacheProvider.class);
-        var listener = mock(AutomationCompositionElementListener.class);
+        var listener = mock(ThreadHandler.class);
         var participantMessagePublisher = mock(ParticipantMessagePublisher.class);
         var ach = new AutomationCompositionHandler(cacheProvider, participantMessagePublisher, listener);
 
@@ -166,13 +165,13 @@ class AutomationCompositionHandlerTest {
         participantDeploy.getAcElementList().add(acElementDeploy);
 
         ach.handleAcPropertyUpdate(updateMsg);
-        verify(listener).update(any(), any(), any());
+        verify(listener).update(any(), any(), any(), any());
     }
 
     @Test
-    void handleAutomationCompositionDeployTest() throws PfModelException {
+    void handleAutomationCompositionDeployTest() {
         var cacheProvider = mock(CacheProvider.class);
-        var listener = mock(AutomationCompositionElementListener.class);
+        var listener = mock(ThreadHandler.class);
         var participantMessagePublisher = mock(ParticipantMessagePublisher.class);
         var ach = new AutomationCompositionHandler(cacheProvider, participantMessagePublisher, listener);
 
@@ -196,27 +195,29 @@ class AutomationCompositionHandlerTest {
             participantDeploy.getAcElementList().add(acElementDeploy);
         }
         ach.handleAutomationCompositionDeploy(deployMsg);
-        verify(listener, times(automationComposition.getElements().size())).deploy(any(), any(), any());
+        verify(listener, times(automationComposition.getElements().size())).deploy(any(), any(), any(), any());
     }
 
     @Test
-    void handleComposiotPrimeTest() throws PfModelException {
-        var listener = mock(AutomationCompositionElementListener.class);
+    void handleComposiotPrimeTest() {
+        var listener = mock(ThreadHandler.class);
         var ach = new AutomationCompositionHandler(mock(CacheProvider.class), mock(ParticipantMessagePublisher.class),
                 listener);
         var compositionId = UUID.randomUUID();
         var list = List.of(new AutomationCompositionElementDefinition());
-        ach.prime(compositionId, list);
-        verify(listener).prime(compositionId, list);
+        var messageId = UUID.randomUUID();
+        ach.prime(messageId, compositionId, list);
+        verify(listener).prime(messageId, compositionId, list);
     }
 
     @Test
-    void handleComposiotDeprimeTest() throws PfModelException {
-        var listener = mock(AutomationCompositionElementListener.class);
+    void handleComposiotDeprimeTest() {
+        var listener = mock(ThreadHandler.class);
         var ach = new AutomationCompositionHandler(mock(CacheProvider.class), mock(ParticipantMessagePublisher.class),
                 listener);
         var compositionId = UUID.randomUUID();
-        ach.deprime(compositionId);
-        verify(listener).deprime(compositionId);
+        var messageId = UUID.randomUUID();
+        ach.deprime(messageId, compositionId);
+        verify(listener).deprime(messageId, compositionId);
     }
 }
diff --git a/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/handler/ParticipantHandlerTest.java b/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/handler/ParticipantHandlerTest.java
index 895d4ed59..237cab224 100644
--- a/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/handler/ParticipantHandlerTest.java
+++ b/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/handler/ParticipantHandlerTest.java
@@ -158,17 +158,20 @@ class ParticipantHandlerTest {
     void handleParticipantPrimeTest() {
         var cacheProvider = mock(CacheProvider.class);
         when(cacheProvider.getParticipantId()).thenReturn(CommonTestData.getParticipantId());
+
+        var participantPrime = new ParticipantPrime();
+        participantPrime.setCompositionId(UUID.randomUUID());
+        participantPrime.setMessageId(UUID.randomUUID());
+        participantPrime.setParticipantDefinitionUpdates(List.of(createParticipantDefinition()));
+
         var publisher = mock(ParticipantMessagePublisher.class);
         var acHandler = mock(AutomationCompositionHandler.class);
         var participantHandler = new ParticipantHandler(acHandler, mock(AutomationCompositionOutHandler.class),
                 publisher, cacheProvider);
 
-        var participantPrime = new ParticipantPrime();
-        participantPrime.setCompositionId(UUID.randomUUID());
-        participantPrime.setParticipantDefinitionUpdates(List.of(createParticipantDefinition()));
         participantHandler.handleParticipantPrime(participantPrime);
         verify(cacheProvider).addElementDefinition(any(), any());
-        verify(acHandler).prime(any(), any());
+        verify(acHandler).prime(any(), any(), any());
     }
 
     @Test
@@ -182,9 +185,11 @@ class ParticipantHandlerTest {
         var participantPrime = new ParticipantPrime();
         var compositionId = UUID.randomUUID();
         participantPrime.setCompositionId(compositionId);
+        var messageId = UUID.randomUUID();
+        participantPrime.setMessageId(messageId);
         participantHandler.handleParticipantPrime(participantPrime);
         verify(cacheProvider).removeElementDefinition(compositionId);
-        verify(acHandler).deprime(compositionId);
+        verify(acHandler).deprime(messageId, compositionId);
     }
 
     @Test
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
new file mode 100644
index 000000000..767a916b5
--- /dev/null
+++ b/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/handler/ThreadHandlerTest.java
@@ -0,0 +1,160 @@
+/*-
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2023 Nordix Foundation.
+ * ================================================================================
+ * Licensed 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.clamp.acm.participant.intermediary.handler;
+
+import static org.mockito.Mockito.clearInvocations;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import javax.ws.rs.core.Response.Status;
+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.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.StateChangeResult;
+import org.onap.policy.models.base.PfModelException;
+
+class ThreadHandlerTest {
+
+    private static final int TIMEOUT = 400;
+
+    @Test
+    void test() throws PfModelException {
+        var listener = mock(AutomationCompositionElementListener.class);
+        var intermediaryApi = mock(ParticipantIntermediaryApi.class);
+        var threadHandler = new ThreadHandler(listener, intermediaryApi, mock(CacheProvider.class));
+
+        var compositionId = UUID.randomUUID();
+        var list = List.of(new AutomationCompositionElementDefinition());
+        var messageId = UUID.randomUUID();
+        threadHandler.prime(messageId, compositionId, list);
+        verify(listener, timeout(TIMEOUT)).prime(compositionId, list);
+
+        clearInvocations(listener);
+        var element = new AcElementDeploy();
+        var elementId = UUID.randomUUID();
+        element.setId(elementId);
+        Map<String, Object> properties = Map.of("key", "value");
+        var instanceId = UUID.randomUUID();
+        threadHandler.deploy(messageId, instanceId, element, properties);
+        verify(listener, timeout(TIMEOUT)).deploy(instanceId, element, properties);
+
+        clearInvocations(listener);
+        threadHandler.update(messageId, instanceId, element, properties);
+        verify(listener, timeout(TIMEOUT)).update(instanceId, element, properties);
+
+        clearInvocations(listener);
+        threadHandler.lock(messageId, instanceId, elementId);
+        verify(listener, timeout(TIMEOUT)).lock(instanceId, elementId);
+
+        clearInvocations(listener);
+        threadHandler.unlock(messageId, instanceId, elementId);
+        verify(listener, timeout(TIMEOUT)).unlock(instanceId, elementId);
+
+        clearInvocations(listener);
+        threadHandler.undeploy(messageId, instanceId, elementId);
+        verify(listener, timeout(TIMEOUT)).undeploy(instanceId, elementId);
+
+        clearInvocations(listener);
+        threadHandler.delete(messageId, instanceId, elementId);
+        verify(listener, timeout(TIMEOUT)).delete(instanceId, elementId);
+
+        clearInvocations(listener);
+        threadHandler.deprime(messageId, compositionId);
+        verify(listener, timeout(TIMEOUT)).deprime(compositionId);
+    }
+
+    @Test
+    void testException() throws PfModelException {
+        var listener = mock(AutomationCompositionElementListener.class);
+        var intermediaryApi = mock(ParticipantIntermediaryApi.class);
+        var threadHandler = new ThreadHandler(listener, intermediaryApi, mock(CacheProvider.class));
+
+        var compositionId = UUID.randomUUID();
+        var list = List.of(new AutomationCompositionElementDefinition());
+        doThrow(new PfModelException(Status.INTERNAL_SERVER_ERROR, "Error")).when(listener).prime(compositionId, list);
+        var messageId = UUID.randomUUID();
+        threadHandler.prime(messageId, compositionId, list);
+        verify(intermediaryApi, timeout(TIMEOUT)).updateCompositionState(compositionId, AcTypeState.COMMISSIONED,
+                StateChangeResult.FAILED, "Composition Defintion prime failed");
+
+        clearInvocations(intermediaryApi);
+        var element = new AcElementDeploy();
+        var elementId = UUID.randomUUID();
+        element.setId(elementId);
+        Map<String, Object> properties = Map.of("key", "value");
+        var instanceId = UUID.randomUUID();
+        doThrow(new PfModelException(Status.INTERNAL_SERVER_ERROR, "Error")).when(listener).deploy(instanceId, element,
+                properties);
+        threadHandler.deploy(messageId, instanceId, element, properties);
+        verify(intermediaryApi, timeout(TIMEOUT)).updateAutomationCompositionElementState(instanceId, elementId,
+                DeployState.UNDEPLOYED, null, StateChangeResult.FAILED, "Automation composition element deploy failed");
+
+        clearInvocations(listener);
+        doThrow(new PfModelException(Status.INTERNAL_SERVER_ERROR, "Error")).when(listener).update(instanceId, element,
+                properties);
+        threadHandler.update(messageId, instanceId, element, properties);
+        verify(intermediaryApi, timeout(TIMEOUT)).updateAutomationCompositionElementState(instanceId, elementId,
+                DeployState.DEPLOYED, null, StateChangeResult.FAILED, "Automation composition element update failed");
+
+        clearInvocations(listener);
+        doThrow(new PfModelException(Status.INTERNAL_SERVER_ERROR, "Error")).when(listener).lock(instanceId, elementId);
+        threadHandler.lock(messageId, instanceId, elementId);
+        verify(intermediaryApi, timeout(TIMEOUT)).updateAutomationCompositionElementState(instanceId, elementId, null,
+                LockState.UNLOCKED, StateChangeResult.FAILED, "Automation composition element lock failed");
+
+        clearInvocations(listener);
+        doThrow(new PfModelException(Status.INTERNAL_SERVER_ERROR, "Error")).when(listener).unlock(instanceId,
+                elementId);
+        threadHandler.unlock(messageId, instanceId, elementId);
+        verify(intermediaryApi, timeout(TIMEOUT)).updateAutomationCompositionElementState(instanceId, elementId, null,
+                LockState.LOCKED, StateChangeResult.FAILED, "Automation composition element unlock failed");
+
+        clearInvocations(listener);
+        doThrow(new PfModelException(Status.INTERNAL_SERVER_ERROR, "Error")).when(listener).undeploy(instanceId,
+                elementId);
+        threadHandler.undeploy(messageId, instanceId, elementId);
+        verify(intermediaryApi, timeout(TIMEOUT)).updateAutomationCompositionElementState(instanceId, elementId,
+                DeployState.DEPLOYED, null, StateChangeResult.FAILED, "Automation composition element undeploy failed");
+
+        clearInvocations(listener);
+        doThrow(new PfModelException(Status.INTERNAL_SERVER_ERROR, "Error")).when(listener).delete(instanceId,
+                elementId);
+        threadHandler.delete(messageId, instanceId, elementId);
+        verify(intermediaryApi, timeout(TIMEOUT)).updateAutomationCompositionElementState(instanceId, elementId,
+                DeployState.UNDEPLOYED, null, StateChangeResult.FAILED, "Automation composition element delete failed");
+
+        clearInvocations(listener);
+        doThrow(new PfModelException(Status.INTERNAL_SERVER_ERROR, "Error")).when(listener).deprime(compositionId);
+        threadHandler.deprime(messageId, compositionId);
+        verify(intermediaryApi, timeout(TIMEOUT)).updateCompositionState(compositionId,
+                AcTypeState.PRIMED, StateChangeResult.FAILED, "Composition Defintion deprime failed");
+    }
+}