2 * ============LICENSE_START=======================================================
4 * ================================================================================
5 * Copyright (C) 2018, 2020 AT&T Intellectual Property. All rights reserved.
6 * ================================================================================
7 * Licensed under the Apache License, Version 2.0 (the "License");
8 * you may not use this file except in compliance with the License.
9 * You may obtain a copy of the License at
11 * http://www.apache.org/licenses/LICENSE-2.0
13 * Unless required by applicable law or agreed to in writing, software
14 * distributed under the License is distributed on an "AS IS" BASIS,
15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * See the License for the specific language governing permissions and
17 * limitations under the License.
18 * ============LICENSE_END=========================================================
21 package org.onap.policy.drools.pooling.state;
23 import org.onap.policy.drools.pooling.PoolingManager;
24 import org.onap.policy.drools.pooling.message.Heartbeat;
25 import org.slf4j.Logger;
26 import org.slf4j.LoggerFactory;
29 * The start state. Upon entry, a heart beat is generated and the event filter is changed
30 * to look for just that particular message. Once the message is seen, it goes into the
33 public class StartState extends State {
35 private static final Logger logger = LoggerFactory.getLogger(StartState.class);
38 * Time stamp inserted into the heart beat message.
40 private long hbTimestampMs = System.currentTimeMillis();
45 * @param mgr pooling manager
47 public StartState(PoolingManager mgr) {
52 * Get Heart beat time stamp in milliseconds.
54 * @return the time stamp inserted into the heart beat message
56 public long getHbTimestampMs() {
65 Heartbeat hb = makeHeartbeat(hbTimestampMs);
66 publish(getHost(), hb);
69 * heart beat generator
71 long genMs = getProperties().getInterHeartbeatMs();
73 scheduleWithFixedDelay(genMs, genMs, () -> {
74 publish(getHost(), hb);
79 * my heart beat checker
81 schedule(getProperties().getStartHeartbeatMs(), () -> {
82 logger.error("missed heartbeat on topic {}", getTopic());
83 return internalTopicFailed();
88 * Transitions to the query state if the heart beat originated from this host and its
92 public State process(Heartbeat msg) {
93 if (msg.getTimestampMs() == hbTimestampMs && getHost().equals(msg.getSource())) {
94 // saw our own heart beat - transition to query state
95 logger.info("saw our own heartbeat on topic {}", getTopic());
100 logger.info("ignored old heartbeat message from {} on topic {}", msg.getSource(), getTopic());