/*-
* ============LICENSE_START=======================================================
- * Copyright (C) 2019 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2019, 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.
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import lombok.Getter;
-import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClient;
+import org.onap.policy.common.endpoints.event.comm.client.BidirectionalTopicClient;
+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.onap.policy.models.pdp.concepts.PdpStatus;
+import org.onap.policy.models.pdp.concepts.PdpTopicCheck;
import org.onap.policy.pdpx.main.XacmlState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class XacmlPdpHearbeatPublisher implements Runnable {
- public static final int DEFAULT_INTERVAL_MS = 60000;
+ public static final int DEFAULT_HB_INTERVAL_MS = 60000;
private static final Logger LOGGER = LoggerFactory.getLogger(XacmlPdpHearbeatPublisher.class);
+ private static final Coder CODER = new StandardCoder();
- private final TopicSinkClient topicSinkClient;
+ private final BidirectionalTopicClient topicChecker;
+ private final long probeHeartbeatTopicMs;
/**
* Tracks the state of this PDP.
* Current timer interval, in milliseconds.
*/
@Getter
- private long intervalMs = DEFAULT_INTERVAL_MS;
+ private long intervalMs = DEFAULT_HB_INTERVAL_MS;
private ScheduledExecutorService timerThread;
/**
* Constructor for instantiating XacmlPdpPublisher.
*
- * @param topicSinkClient used to send heart beat message
+ * @param topicChecker used to check the topic before sending heart beat message
+ * @param probeHeartbeatTopicMs frequency, in milliseconds, with which to probe the
+ * heartbeat topic before sending the first heartbeat. Zero disables probing
* @param state tracks the state of this PDP
*/
- public XacmlPdpHearbeatPublisher(TopicSinkClient topicSinkClient, XacmlState state) {
- this.topicSinkClient = topicSinkClient;
+ public XacmlPdpHearbeatPublisher(BidirectionalTopicClient topicChecker, long probeHeartbeatTopicMs,
+ XacmlState state) {
+ LOGGER.info("heartbeat topic probe {}ms", probeHeartbeatTopicMs);
+ this.topicChecker = topicChecker;
+ this.probeHeartbeatTopicMs = probeHeartbeatTopicMs;
this.currentState = state;
}
@Override
public void run() {
- PdpStatus message = currentState.genHeartbeat();
- LOGGER.info("Sending Xacml PDP heartbeat to the PAP - {}", message);
+ try {
+ if (!isTopicReady()) {
+ return;
+ }
+
+ PdpStatus message = currentState.genHeartbeat();
+ LOGGER.info("Sending Xacml PDP heartbeat to the PAP - {}", message);
+
+ String json = CODER.encode(message);
+ topicChecker.send(json);
- topicSinkClient.send(message);
+ } catch (RuntimeException | CoderException e) {
+ LOGGER.warn("send to {} failed because of {}", topicChecker.getSink().getTopic(), e.getMessage(), e);
+ }
+ }
+
+ private boolean isTopicReady() throws CoderException {
+ if (probeHeartbeatTopicMs <= 0 || topicChecker.isReady()) {
+ return true;
+ }
+
+ var check = new PdpTopicCheck();
+ check.setName(XacmlState.PDP_NAME);
+ return topicChecker.awaitReady(check, probeHeartbeatTopicMs);
}
/**
* Method to terminate the heart beat.
*/
public synchronized void terminate() {
+ topicChecker.stopWaiting();
+
if (timerThread != null) {
timerThread.shutdownNow();
timerThread = null;
package org.onap.policy.pdpx.main.parameters;
import lombok.Getter;
+import lombok.NoArgsConstructor;
import org.apache.commons.lang3.StringUtils;
import org.onap.policy.common.endpoints.parameters.RestClientParameters;
import org.onap.policy.common.endpoints.parameters.RestServerParameters;
import org.onap.policy.common.parameters.BeanValidationResult;
import org.onap.policy.common.parameters.ParameterGroupImpl;
import org.onap.policy.common.parameters.ValidationStatus;
+import org.onap.policy.common.parameters.annotations.Min;
import org.onap.policy.common.parameters.annotations.NotBlank;
import org.onap.policy.common.parameters.annotations.NotNull;
import org.onap.policy.common.parameters.annotations.Valid;
@Getter
@NotNull
@NotBlank
+@NoArgsConstructor
public class XacmlPdpParameterGroup extends ParameterGroupImpl {
public static final String PARAM_POLICY_API = "policyApiParameters";
private TopicParameterGroup topicParameterGroup;
@Valid
private XacmlApplicationParameters applicationParameters;
+ /**
+ * Frequency, in seconds, with which to probe the heartbeat topic before sending the
+ * first heartbeat. Set to zero to disable probing.
+ */
+ @Min(0)
+ private long probeHeartbeatTopicSec = 4;
+
/**
* Create the xacml pdp parameter group.
package org.onap.policy.pdpx.main.startstop;
-import java.util.Arrays;
import lombok.Getter;
import lombok.Setter;
import org.onap.policy.common.endpoints.event.comm.TopicEndpointManager;
-import org.onap.policy.common.endpoints.event.comm.TopicSource;
+import org.onap.policy.common.endpoints.event.comm.client.BidirectionalTopicClient;
+import org.onap.policy.common.endpoints.event.comm.client.BidirectionalTopicClientException;
import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClient;
-import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClientException;
import org.onap.policy.common.endpoints.http.client.HttpClient;
import org.onap.policy.common.endpoints.http.client.HttpClientConfigException;
import org.onap.policy.common.endpoints.http.client.HttpClientFactoryInstance;
// The parameters of this policy xacml pdp activator
private final XacmlPdpParameterGroup xacmlPdpParameterGroup;
+ /**
+ * POLICY-PDP-PAP client.
+ */
+ private BidirectionalTopicClient topicClient;
+
/**
* Listens for messages on the topic, decodes them into a {@link PdpStatus} message, and then
* dispatches them to appropriate listener.
this.xacmlPdpParameterGroup = xacmlPdpParameterGroup;
this.msgDispatcher = new MessageTypeDispatcher(MSG_TYPE_NAMES);
- sinkClient = new TopicSinkClient(TOPIC);
- heartbeat = new XacmlPdpHearbeatPublisher(sinkClient, state);
+ topicClient = new BidirectionalTopicClient(TOPIC, TOPIC);
+ sinkClient = new TopicSinkClient(topicClient.getSink());
+
+ heartbeat = new XacmlPdpHearbeatPublisher(topicClient,
+ xacmlPdpParameterGroup.getProbeHeartbeatTopicSec() * 1000, state);
/*
* since the dispatcher isn't registered with the topic yet, we can go ahead
restServer = new XacmlPdpRestServer(xacmlPdpParameterGroup.getRestServerParameters(),
XacmlPdpAafFilter.class, XacmlPdpRestController.class);
- } catch (RuntimeException | TopicSinkClientException | HttpClientConfigException e) {
+ } catch (RuntimeException | HttpClientConfigException | BidirectionalTopicClientException e) {
throw new PolicyXacmlPdpRuntimeException(e.getMessage(), e);
}
* Registers the dispatcher with the topic source(s).
*/
private void registerMsgDispatcher() {
- for (TopicSource source : TopicEndpointManager.getManager().getTopicSources(Arrays.asList(TOPIC))) {
- source.register(msgDispatcher);
- }
+ topicClient.getSource().register(msgDispatcher);
}
/**
* Unregisters the dispatcher from the topic source(s).
*/
private void unregisterMsgDispatcher() {
- for (TopicSource source : TopicEndpointManager.getManager().getTopicSources(Arrays.asList(TOPIC))) {
- source.unregister(msgDispatcher);
- }
+ topicClient.getSource().unregister(msgDispatcher);
}
/**
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
-import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClient;
+import org.onap.policy.common.endpoints.event.comm.TopicSink;
+import org.onap.policy.common.endpoints.event.comm.client.BidirectionalTopicClient;
+import org.onap.policy.common.utils.coder.CoderException;
import org.onap.policy.models.pdp.concepts.PdpStatus;
import org.onap.policy.pdpx.main.XacmlState;
private static final long INTERVAL_INVALID = 0;
@Mock
- private TopicSinkClient client;
+ private TopicSink sink;
+
+ @Mock
+ private BidirectionalTopicClient checker;
@Mock
private XacmlState state;
@Mock
private ScheduledFuture<?> timer2;
- @Mock
private PdpStatus status;
private Queue<ScheduledFuture<?>> timers;
*/
@Before
public void setUp() {
+ when(sink.getTopic()).thenReturn("my-topic");
+ when(checker.getSink()).thenReturn(sink);
+ when(checker.isReady()).thenReturn(true);
when(state.genHeartbeat()).thenReturn(status);
+ status = new PdpStatus();
timers = new LinkedList<>(Arrays.asList(timer1, timer2));
when(executor.scheduleWithFixedDelay(any(), anyLong(), anyLong(), any())).thenAnswer(args -> timers.remove());
- publisher = new MyPublisher(client, state);
+ publisher = new MyPublisher(checker, 10, state);
}
@Test
publisher.run();
verify(state).genHeartbeat();
- verify(client).send(status);
+ verify(checker).send(any());
+ }
+
+ /**
+ * Tests the run() method when the probe is disabled.
+ */
+ @Test
+ public void testRunNoProbe() throws CoderException {
+ publisher = new MyPublisher(checker, 0, state);
+
+ publisher.run();
+
+ verify(checker, never()).isReady();
+ verify(checker, never()).awaitReady(any(), anyLong());
+
+ verify(state).genHeartbeat();
+ verify(checker).send(any());
+ }
+
+ /**
+ * Tests the run() method when the topic is not ready, and then becomes ready.
+ */
+ @Test
+ public void testRunNotReady() throws CoderException {
+ // not ready yet
+ when(checker.isReady()).thenReturn(false);
+ when(checker.awaitReady(any(), anyLong())).thenReturn(false);
+
+ publisher.run();
+ verify(state, never()).genHeartbeat();
+ verify(checker, never()).send(any());
+
+ // isReady is still false, but awaitReady is now true - should generate heartbeat
+ when(checker.awaitReady(any(), anyLong())).thenReturn(true);
+
+ publisher.run();
+ verify(state).genHeartbeat();
+ verify(checker).send(any());
+
+ // now isReady is true, too - should not rerun awaitReady
+ when(checker.isReady()).thenReturn(true);
+
+ publisher.run();
+ verify(state, times(2)).genHeartbeat();
+ verify(checker, times(2)).send(any());
+ verify(checker, times(2)).awaitReady(any(), anyLong());
+ }
+
+ /**
+ * Tests the run() method when the checker throws an exception.
+ */
+ @Test
+ public void testRunCheckerEx() throws CoderException {
+ // force it to call awaitReady
+ when(checker.isReady()).thenReturn(false);
+
+ when(checker.awaitReady(any(), anyLong()))
+ .thenThrow(new CoderException("expected exception"))
+ .thenReturn(true);
+
+ // exception thrown - should not generate heartbeat
+ publisher.run();
+ verify(state, never()).genHeartbeat();
+ verify(checker, never()).send(any());
+
+ // no exception this time - SHOULD generate heartbeat
+ publisher.run();
+ verify(state).genHeartbeat();
+ verify(checker).send(any());
}
@Test
// not yet started
publisher.terminate();
+ verify(checker).stopWaiting();
+
// now start it and then try again
publisher.start();
public void testStart() {
publisher.start();
- verify(executor).scheduleWithFixedDelay(publisher, 0, XacmlPdpHearbeatPublisher.DEFAULT_INTERVAL_MS,
+ verify(executor).scheduleWithFixedDelay(publisher, 0, XacmlPdpHearbeatPublisher.DEFAULT_HB_INTERVAL_MS,
TimeUnit.MILLISECONDS);
// repeat - nothing more should happen
@Test
public void testMakeTimerThread() {
// create a plain listener to test the "real" makeTimer() method
- publisher = new XacmlPdpHearbeatPublisher(client, state);
+ publisher = new XacmlPdpHearbeatPublisher(checker, 1, state);
assertThatCode(() -> {
publisher.start();
private class MyPublisher extends XacmlPdpHearbeatPublisher {
- public MyPublisher(TopicSinkClient topicSinkClient, XacmlState state) {
- super(topicSinkClient, state);
+ public MyPublisher(BidirectionalTopicClient topicChecker, long probeHeartbeatTopicMs, XacmlState state) {
+ super(topicChecker, probeHeartbeatTopicMs, state);
}
@Override
import org.onap.policy.pdpx.main.parameters.CommonTestData;
import org.onap.policy.pdpx.main.parameters.XacmlPdpParameterGroup;
import org.onap.policy.pdpx.main.parameters.XacmlPdpParameterHandler;
+import org.powermock.reflect.Whitebox;
/**
*
*/
public class TestXacmlPdpActivator extends CommonRest {
+ private static final String PROBE_FIELD_NAME = "probeHeartbeatTopicSec";
+
private static XacmlPdpParameterGroup parGroup;
private XacmlPdpActivator activator = null;
@Override
@Before
public void setUp() {
+ Whitebox.setInternalState(parGroup, PROBE_FIELD_NAME, 4);
activator = new XacmlPdpActivator(parGroup);
}
assertTrue(activator.isXacmlRestControllerAlive());
}
+ @Test
+ public void testXacmlPdpActivator_NoProbe() throws Exception {
+ Whitebox.setInternalState(parGroup, PROBE_FIELD_NAME, 0);
+ activator = new XacmlPdpActivator(parGroup);
+ activator.start();
+ assertTrue(activator.isAlive());
+ }
+
@Test
public void testGetCurrent_testSetCurrent() {
XacmlPdpActivator.setCurrent(activator);
Code Review / policy / xacml-pdp.git / commitdiff