e05b3f1baeefc880f50c651ed43ab71f02c304be
[ccsdk/features.git] /
1 /*
2  * ============LICENSE_START=======================================================
3  * ONAP : ccsdk features
4  * ================================================================================
5  * Copyright (C) 2019 highstreet technologies GmbH Intellectual Property.
6  * All rights reserved.
7  * ================================================================================
8  * Update Copyright (C) 2020 AT&T Intellectual Property. All rights reserved.
9  * ================================================================================
10  * Licensed under the Apache License, Version 2.0 (the "License");
11  * you may not use this file except in compliance with the License.
12  * You may obtain a copy of the License at
13  *
14  *     http://www.apache.org/licenses/LICENSE-2.0
15  *
16  * Unless required by applicable law or agreed to in writing, software
17  * distributed under the License is distributed on an "AS IS" BASIS,
18  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
19  * See the License for the specific language governing permissions and
20  * limitations under the License.
21  * ============LICENSE_END=======================================================
22  *
23  */
24 package org.onap.ccsdk.features.sdnr.wt.mountpointstateprovider.impl;
25
26 import java.util.LinkedList;
27 import java.util.List;
28 import org.eclipse.jdt.annotation.NonNull;
29 import org.json.JSONObject;
30 import org.onap.ccsdk.features.sdnr.wt.devicemanager.service.VESCollectorCfgService;
31 import org.onap.ccsdk.features.sdnr.wt.devicemanager.service.VESCollectorService;
32 import org.slf4j.Logger;
33 import org.slf4j.LoggerFactory;
34
35 public class MountpointStatePublisher implements Runnable {
36
37     private static final Logger LOG = LoggerFactory.getLogger(MountpointStatePublisher.class);
38     private List<JSONObject> stateObjects = new LinkedList<>();
39     private boolean publish = true;
40     private int publishPause = 5000; // Default pause between fetch - 5 seconds
41     private VESCollectorService vesCollectorService;
42
43     public MountpointStatePublisher(@NonNull VESCollectorService vesCollectorService) {
44         this.vesCollectorService = vesCollectorService;
45     }
46
47     public void addToPublish(JSONObject publishObj) {
48         getStateObjects().add(publishObj);
49     }
50
51     public synchronized List<JSONObject> getStateObjects() {
52         return stateObjects;
53     }
54
55     public void stop() {
56         publish = false;
57     }
58
59     private void pauseThread() throws InterruptedException {
60         if (publishPause > 0) {
61             LOG.debug("No data yet to publish.  Pausing {} ms before retry ", publishPause);
62             Thread.sleep(publishPause);
63         } else {
64             LOG.debug("No data yet to publish. No publish pause specified - retrying immediately");
65         }
66     }
67
68
69     public String createVESMessage(JSONObject msg, VESCollectorCfgService vesCfg) {
70         MountpointStateVESMessageFormatter vesFormatter = new MountpointStateVESMessageFormatter(vesCfg);
71         return vesFormatter.createVESMessage(msg);
72     }
73
74     @Override
75     public void run() {
76         while (publish) {
77             try {
78                 if (!getStateObjects().isEmpty()) {
79                     JSONObject obj = ((LinkedList<JSONObject>) getStateObjects()).removeFirst();
80                     String vesMsg = createVESMessage(obj, vesCollectorService.getConfig());
81                     this.vesCollectorService.publishVESMessage(vesMsg);
82                 } else {
83                     pauseThread();
84                 }
85             } catch (Exception ex) {
86                 LOG.error("Exception while publishing message, ignoring and continuing ... ", ex);
87             }
88         }
89     }
90 }