aboutsummaryrefslogtreecommitdiffstats
path: root/policy-endpoints/src/main
diff options
context:
space:
mode:
authorJim Hahn <jrh3@att.com>2021-08-18 10:46:28 -0400
committerJim Hahn <jrh3@att.com>2021-08-18 16:20:40 -0400
commit5d79140dbdaa85c390c3e8f6914739c2c8a6b130 (patch)
treec1eec29103c67232fbbb570ee3a903bcea3e5b0b /policy-endpoints/src/main
parentecc85ec4c6af67dc1ba2ff20b7473925e6657b11 (diff)
Add topic checker
Added a method to check the readiness of a bidirectional topic. Issue-ID: POLICY-3531 Change-Id: I2fefae7ba1ea5ed9ed33140717d05828e6dec94d Signed-off-by: Jim Hahn <jrh3@att.com>
Diffstat (limited to 'policy-endpoints/src/main')
-rw-r--r--policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/client/BidirectionalTopicClient.java117
1 files changed, 116 insertions, 1 deletions
diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/client/BidirectionalTopicClient.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/client/BidirectionalTopicClient.java
index 57a331c1..2a9f1446 100644
--- a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/client/BidirectionalTopicClient.java
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/client/BidirectionalTopicClient.java
@@ -2,7 +2,7 @@
* ============LICENSE_START=======================================================
* ONAP
* ================================================================================
- * Copyright (C) 2020 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2020-2021 AT&T Intellectual Property. All rights reserved.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -22,6 +22,9 @@ package org.onap.policy.common.endpoints.event.comm.client;
import java.util.Arrays;
import java.util.List;
+import java.util.concurrent.BlockingDeque;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.TimeUnit;
import lombok.Getter;
import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
import org.onap.policy.common.endpoints.event.comm.TopicEndpoint;
@@ -29,6 +32,11 @@ import org.onap.policy.common.endpoints.event.comm.TopicEndpointManager;
import org.onap.policy.common.endpoints.event.comm.TopicListener;
import org.onap.policy.common.endpoints.event.comm.TopicSink;
import org.onap.policy.common.endpoints.event.comm.TopicSource;
+import org.onap.policy.common.utils.coder.Coder;
+import org.onap.policy.common.utils.coder.CoderException;
+import org.onap.policy.common.utils.coder.StandardCoder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* A "bidirectional" topic, which is a pair of topics, one of which is used to publish
@@ -36,6 +44,9 @@ import org.onap.policy.common.endpoints.event.comm.TopicSource;
*/
@Getter
public class BidirectionalTopicClient {
+ private static final Logger logger = LoggerFactory.getLogger(BidirectionalTopicClient.class);
+ private static final Coder coder = new StandardCoder();
+
private final String sinkTopic;
private final String sourceTopic;
private final TopicSink sink;
@@ -44,6 +55,16 @@ public class BidirectionalTopicClient {
private final CommInfrastructure sourceTopicCommInfrastructure;
/**
+ * Used when checking whether or not a message sent on the sink topic can be received
+ * on the source topic. When a matching message is received on the incoming topic,
+ * {@code true} is placed on the queue. If {@link #stop()} is called or the waiting
+ * thread is interrupted, then {@code false} is placed on the queue. Whenever a value
+ * is pulled from the queue, it is immediately placed back on the queue.
+ */
+ private final BlockingDeque<Boolean> checkerQueue = new LinkedBlockingDeque<>();
+
+
+ /**
* Constructs the object.
*
* @param sinkTopic sink topic name
@@ -94,9 +115,103 @@ public class BidirectionalTopicClient {
source.unregister(topicListener);
}
+ /**
+ * Determines whether or not the topic is ready (i.e., {@link #awaitReady(Object)} has
+ * previously returned {@code true}).
+ *
+ * @return {@code true}, if the topic is ready to send and receive
+ */
+ public boolean isReady() {
+ return Boolean.TRUE.equals(checkerQueue.peek());
+ }
+
+ /**
+ * Waits for the bidirectional topic to become "ready" by publishing a message on the
+ * sink topic and awaiting receipt of the message on the source topic. If the message
+ * is not received within a few seconds, then it tries again. This process is
+ * continued until the message is received, {@link #stop()} is called, or this thread
+ * is interrupted. Once this returns, subsequent calls will return immediately, always
+ * with the same value.
+ *
+ * @param message message to be sent to the sink topic. Note: the equals() method must
+ * return {@code true} if and only if two messages are the same
+ * @param waitMs time to wait, in milliseconds, before re-sending the message
+ * @return {@code true} if the message was received from the source topic,
+ * {@code false} if this method was stopped or interrupted before receipt of
+ * the message
+ * @throws CoderException if the message cannot be encoded
+ */
+ public synchronized <T> boolean awaitReady(T message, long waitMs) throws CoderException {
+ // see if we already know the answer
+ if (!checkerQueue.isEmpty()) {
+ return checkerQueue.peek();
+ }
+
+ final String messageText = coder.encode(message);
+
+ // class of message to be decoded
+ @SuppressWarnings("unchecked")
+ final Class<? extends T> clazz = (Class<? extends T>) message.getClass();
+
+ // create a listener to detect when a matching message is received
+ final TopicListener listener = (infra, topic, msg) -> {
+ try {
+ T incoming = decode(msg, clazz);
+
+ if (message.equals(incoming)) {
+ logger.info("topic {} is ready; found matching message {}", topic, incoming);
+ checkerQueue.add(Boolean.TRUE);
+ }
+
+ } catch (CoderException e) {
+ logger.warn("cannot decode message from topic {}", topic, e);
+ decodeFailed();
+ }
+ };
+
+ source.register(listener);
+
+ // loop until the message is received
+ try {
+ Boolean result;
+ do {
+ send(messageText);
+ } while ((result = checkerQueue.poll(waitMs, TimeUnit.MILLISECONDS)) == null);
+
+ // put it back on the queue
+ checkerQueue.add(result);
+
+ } catch (InterruptedException e) {
+ logger.error("interrupted waiting for topic sink {} source {}", sink.getTopic(), source.getTopic(), e);
+ Thread.currentThread().interrupt();
+ checkerQueue.add(Boolean.FALSE);
+
+ } finally {
+ source.unregister(listener);
+ }
+
+ return checkerQueue.peek();
+ }
+
+ /**
+ * Stops any listeners that are currently stuck in {@link #awaitReady(Object)} by
+ * adding {@code false} to the queue.
+ */
+ public void stopWaiting() {
+ checkerQueue.add(Boolean.FALSE);
+ }
+
// these may be overridden by junit tests
protected TopicEndpoint getTopicEndpointManager() {
return TopicEndpointManager.getManager();
}
+
+ protected <T> T decode(String msg, Class<? extends T> clazz) throws CoderException {
+ return coder.decode(msg, clazz);
+ }
+
+ protected void decodeFailed() {
+ // already logged - nothing else to do
+ }
}