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>

4 files changed, 88 insertions, 19 deletions

diff --git a/models/src/main/java/org/onap/policy/clamp/controlloop/models/messages/dmaap/participant/ParticipantAckMessage.java b/models/src/main/java/org/onap/policy/clamp/controlloop/models/messages/dmaap/participant/ParticipantAckMessage.java
index 8b59a1801..c6f5c61b9 100644
--- a/models/src/main/java/org/onap/policy/clamp/controlloop/models/messages/dmaap/participant/ParticipantAckMessage.java
+++ b/models/src/main/java/org/onap/policy/clamp/controlloop/models/messages/dmaap/participant/ParticipantAckMessage.java
@@ -22,6 +22,7 @@ package org.onap.policy.clamp.controlloop.models.messages.dmaap.participant;
 
 import java.util.UUID;
 import lombok.Getter;
+import lombok.NonNull;
 import lombok.Setter;
 import lombok.ToString;
 import org.onap.policy.models.tosca.authorative.concepts.ToscaConceptIdentifier;
@@ -77,4 +78,31 @@ public class ParticipantAckMessage {
         this.participantType = source.participantType;
         this.participantId = source.participantId;
     }
+
+    /**
+     * Determines if this message applies to this participant type.
+     *
+     * @param participantType type of the participant to match against
+     * @param participantId id of the participant to match against
+     * @return {@code true} if this message applies to this participant, {@code false} otherwise
+     */
+    public boolean appliesTo(@NonNull final ToscaConceptIdentifier participantType,
+            @NonNull final ToscaConceptIdentifier participantId) {
+        // Broadcast message to all participants
+        if (this.participantType == null) {
+            return true;
+        }
+
+        if (!participantType.equals(this.participantType)) {
+            return false;
+        }
+
+        // Broadcast message to all control loop elements on this participant
+        if (this.participantId == null) {
+            return true;
+        }
+
+        // Targeted message at this specific participant
+        return participantId.equals(this.participantId);
+    }
 }
diff --git a/models/src/main/java/org/onap/policy/clamp/controlloop/models/messages/dmaap/participant/ParticipantMessage.java b/models/src/main/java/org/onap/policy/clamp/controlloop/models/messages/dmaap/participant/ParticipantMessage.java
index 3ca4d3d34..f98a88c3b 100644
--- a/models/src/main/java/org/onap/policy/clamp/controlloop/models/messages/dmaap/participant/ParticipantMessage.java
+++ b/models/src/main/java/org/onap/policy/clamp/controlloop/models/messages/dmaap/participant/ParticipantMessage.java
@@ -97,17 +97,16 @@ public class ParticipantMessage {
             return true;
         }
 
-        // Broadcast message to all control loop elements on this participant
-        if (participantType.equals(this.participantType) && this.participantId == null) {
-            return true;
+        if (!participantType.equals(this.participantType)) {
+            return false;
         }
 
-        // Targeted message at this specific participant
-        if (participantType.equals(this.participantType) && participantId.equals(this.participantId)) {
+        // Broadcast message to all control loop elements on this participant
+        if (this.participantId == null) {
             return true;
         }
 
-        // Message is not for this participant
-        return false;
+        // Targeted message at this specific participant
+        return participantId.equals(this.participantId);
     }
 }
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));