Fix issue in event handling in participants

Fix issue in event handling in participants
and refactor Participant Publisher and Listener

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

2 files changed, 54 insertions, 12 deletions

diff --git a/models/src/test/java/org/onap/policy/clamp/controlloop/models/messages/dmaap/participant/ParticipantAckMessageTest.java b/models/src/test/java/org/onap/policy/clamp/controlloop/models/messages/dmaap/participant/ParticipantAckMessageTest.java
index b9c1053f3..df82ab071 100644
--- a/models/src/test/java/org/onap/policy/clamp/controlloop/models/messages/dmaap/participant/ParticipantAckMessageTest.java
+++ b/models/src/test/java/org/onap/policy/clamp/controlloop/models/messages/dmaap/participant/ParticipantAckMessageTest.java
@@ -22,15 +22,23 @@ package org.onap.policy.clamp.controlloop.models.messages.dmaap.participant;
 
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 import static org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantMessageUtils.assertSerializable;
 
 import java.util.UUID;
 import org.junit.jupiter.api.Test;
 import org.onap.policy.common.utils.coder.CoderException;
+import org.onap.policy.models.tosca.authorative.concepts.ToscaConceptIdentifier;
 
 class ParticipantAckMessageTest {
     private ParticipantAckMessage message;
 
+    private static final ToscaConceptIdentifier PTYPE_456 = new ToscaConceptIdentifier("PType", "4.5.6");
+    private static final ToscaConceptIdentifier PTYPE_457 = new ToscaConceptIdentifier("PType", "4.5.7");
+    private static final ToscaConceptIdentifier ID_123 = new ToscaConceptIdentifier("id", "1.2.3");
+    private static final ToscaConceptIdentifier ID_124 = new ToscaConceptIdentifier("id", "1.2.4");
+
     @Test
     void testCopyConstructor() throws CoderException {
         assertThatThrownBy(() -> new ParticipantAckMessage((ParticipantAckMessage) null))
@@ -51,9 +59,43 @@ class ParticipantAckMessageTest {
         assertSerializable(message, ParticipantAckMessage.class);
     }
 
+    @Test
+    void testAppliesTo_NullParticipantId() {
+        message = makeMessage();
+
+        assertThatThrownBy(() -> message.appliesTo(null, null)).isInstanceOf(NullPointerException.class);
+        assertThatThrownBy(() -> message.appliesTo(PTYPE_456, null)).isInstanceOf(NullPointerException.class);
+        assertThatThrownBy(() -> message.appliesTo(null, ID_123)).isInstanceOf(NullPointerException.class);
+    }
+
+    @Test
+    void testAppliesTo_ParticipantIdMatches() {
+        message = makeMessage();
+
+        // ParticipantId matches
+        assertTrue(message.appliesTo(PTYPE_456, ID_123));
+        assertFalse(message.appliesTo(PTYPE_456, ID_124));
+        assertFalse(message.appliesTo(PTYPE_457, ID_123));
+    }
+
+    @Test
+    void testAppliesTo_ParticipantIdNoMatch() {
+        message = makeMessage();
+
+        // ParticipantId does not match
+        ToscaConceptIdentifier id = new ToscaConceptIdentifier();
+        id.setName("id1111");
+        id.setVersion("3.2.1");
+        assertFalse(message.appliesTo(id, id));
+        message.setParticipantType(null);
+        assertTrue(message.appliesTo(id, id));
+    }
+
     private ParticipantAckMessage makeMessage() {
         ParticipantAckMessage msg = new ParticipantAckMessage(ParticipantMessageType.PARTICIPANT_DEREGISTER_ACK);
 
+        msg.setParticipantType(PTYPE_456);
+        msg.setParticipantId(ID_123);
         msg.setMessage("Successfull Ack");
         msg.setResult(true);
         msg.setResponseTo(UUID.randomUUID());
diff --git a/models/src/test/java/org/onap/policy/clamp/controlloop/models/messages/dmaap/participant/ParticipantMessageTest.java b/models/src/test/java/org/onap/policy/clamp/controlloop/models/messages/dmaap/participant/ParticipantMessageTest.java
index 924ad8fa8..58d3afebc 100644
--- a/models/src/test/java/org/onap/policy/clamp/controlloop/models/messages/dmaap/participant/ParticipantMessageTest.java
+++ b/models/src/test/java/org/onap/policy/clamp/controlloop/models/messages/dmaap/participant/ParticipantMessageTest.java
@@ -35,6 +35,11 @@ import org.onap.policy.models.tosca.authorative.concepts.ToscaConceptIdentifier;
 class ParticipantMessageTest {
     private ParticipantMessage message;
 
+    private static final ToscaConceptIdentifier PTYPE_456 = new ToscaConceptIdentifier("PType", "4.5.6");
+    private static final ToscaConceptIdentifier PTYPE_457 = new ToscaConceptIdentifier("PType", "4.5.7");
+    private static final ToscaConceptIdentifier ID_123 = new ToscaConceptIdentifier("id", "1.2.3");
+    private static final ToscaConceptIdentifier ID_124 = new ToscaConceptIdentifier("id", "1.2.4");
+
     @Test
     void testCopyConstructor() throws CoderException {
         assertThatThrownBy(() -> new ParticipantMessage((ParticipantMessage) null))
@@ -62,10 +67,8 @@ class ParticipantMessageTest {
         message = makeMessage();
 
         assertThatThrownBy(() -> message.appliesTo(null, null)).isInstanceOf(NullPointerException.class);
-        assertThatThrownBy(() -> message.appliesTo(new ToscaConceptIdentifier("PType", "4.5.6"), null))
-                .isInstanceOf(NullPointerException.class);
-        assertThatThrownBy(() -> message.appliesTo(null, new ToscaConceptIdentifier("id", "1.2.3")))
-                .isInstanceOf(NullPointerException.class);
+        assertThatThrownBy(() -> message.appliesTo(PTYPE_456, null)).isInstanceOf(NullPointerException.class);
+        assertThatThrownBy(() -> message.appliesTo(null, ID_123)).isInstanceOf(NullPointerException.class);
     }
 
     @Test
@@ -73,12 +76,9 @@ class ParticipantMessageTest {
         message = makeMessage();
 
         // ParticipantId matches
-        assertTrue(message.appliesTo(new ToscaConceptIdentifier("PType", "4.5.6"),
-                new ToscaConceptIdentifier("id", "1.2.3")));
-        assertFalse(message.appliesTo(new ToscaConceptIdentifier("PType", "4.5.6"),
-                new ToscaConceptIdentifier("id", "1.2.4")));
-        assertFalse(message.appliesTo(new ToscaConceptIdentifier("PType", "4.5.7"),
-                new ToscaConceptIdentifier("id", "1.2.3")));
+        assertTrue(message.appliesTo(PTYPE_456, ID_123));
+        assertFalse(message.appliesTo(PTYPE_456, ID_124));
+        assertFalse(message.appliesTo(PTYPE_457, ID_123));
     }
 
     @Test
@@ -97,8 +97,8 @@ class ParticipantMessageTest {
     private ParticipantMessage makeMessage() {
         ParticipantMessage msg = new ParticipantMessage(ParticipantMessageType.PARTICIPANT_STATE_CHANGE);
 
-        msg.setParticipantType(new ToscaConceptIdentifier("PType", "4.5.6"));
-        msg.setParticipantId(new ToscaConceptIdentifier("id", "1.2.3"));
+        msg.setParticipantType(PTYPE_456);
+        msg.setParticipantId(ID_123);
         msg.setMessageId(UUID.randomUUID());
         msg.setTimestamp(Instant.ofEpochMilli(3000));