sdnr/wt/mountpoint-state-provider/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/mountpointstateprovider/impl/MountpointStatePublisher.java

   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
  25 package org.onap.ccsdk.features.sdnr.wt.mountpointstateprovider.impl;
  26
  27 import java.io.IOException;
  28 import java.util.LinkedList;
  29 import java.util.List;
  30 import java.util.Properties;
  31 import java.util.concurrent.TimeUnit;
  32
  33 import org.json.JSONObject;
  34 import org.onap.ccsdk.features.sdnr.wt.common.configuration.Configuration;
  35 import org.onap.dmaap.mr.client.MRBatchingPublisher;
  36 import org.onap.dmaap.mr.client.MRClientFactory;
  37 import org.onap.dmaap.mr.client.response.MRPublisherResponse;
  38 import org.slf4j.Logger;
  39 import org.slf4j.LoggerFactory;
  40
  41
  42 public class MountpointStatePublisher implements Runnable {
  43
  44     private static final Logger LOG = LoggerFactory.getLogger(MountpointStatePublisher.class);
  45     public static final List<JSONObject> stateObjects = new LinkedList<JSONObject>();
  46     static MRBatchingPublisher pub;
  47     Properties publisherProperties = new Properties();
  48     static boolean closePublisher = false; //Set this to true in the "Close" method of MountpointStateProviderImpl
  49     private int fetchPause = 5000; // Default pause between fetch - 5 seconds
  50
  51
  52     public MountpointStatePublisher(Configuration config) {
  53         initialize(config);
  54     }
  55
  56     public void initialize(Configuration config) {
  57         LOG.info("In initializePublisher method of MountpointStatePublisher");
  58         GeneralConfig generalCfg = (GeneralConfig) config;
  59
  60         publisherProperties.put(GeneralConfig.PROPERTY_KEY_PUBLISHER_TRANSPORTTYPE, generalCfg.getTransportType());
  61         publisherProperties.put(GeneralConfig.PROPERTY_KEY_PUBLISHER_HOST_PORT, generalCfg.getHostPort());
  62         publisherProperties.put(GeneralConfig.PROPERTY_KEY_PUBLISHER_CONTENTTYPE, generalCfg.getContenttype());
  63         publisherProperties.put(GeneralConfig.PROPERTY_KEY_PUBLISHER_TOPIC, generalCfg.getTopic());
  64         publisherProperties.put(GeneralConfig.PROPERTY_KEY_PUBLISHER_MAXBATCHSIZE, generalCfg.getMaxBatchSize());
  65         publisherProperties.put(GeneralConfig.PROPERTY_KEY_PUBLISHER_MAXAGEMS, generalCfg.getMaxAgeMs());
  66         publisherProperties.put(GeneralConfig.PROPERTY_KEY_PUBLISHER_MESSAGESENTTHREADOCCURANCE,
  67                 generalCfg.getMessageSentThreadOccurrence());
  68
  69         createPublisher(publisherProperties);
  70     }
  71
  72     public MRBatchingPublisher createPublisher(Properties publisherProperties) {
  73
  74         try {
  75             pub = MRClientFactory.createBatchingPublisher(publisherProperties, false);
  76             return pub;
  77         } catch (IOException e) {
  78             LOG.info("Exception while creating a publisher", e);
  79
  80         }
  81         return null;
  82     }
  83
  84     public void publishMessage(MRBatchingPublisher pub, String msg) {
  85         LOG.info("Publishing message {} - ", msg);
  86         try {
  87             pub.send(msg);
  88         } catch (IOException e) {
  89             LOG.info("Exception while publishing a mesage ", e);
  90         }
  91     }
  92
  93     public MRBatchingPublisher getPublisher() {
  94         return pub;
  95     }
  96
  97     public void run() {
  98
  99         while (!closePublisher) {
 100             try {
 101                 if (stateObjects.size() > 0) {
 102                     JSONObject obj = ((LinkedList<JSONObject>) stateObjects).removeFirst();
 103                     publishMessage(getPublisher(), obj.toString());
 104                 } else {
 105                     pauseThread();
 106                 }
 107             } catch (Exception ex) {
 108                 LOG.error("Exception while publishing message, ignoring and continuing ... ", ex);
 109             }
 110
 111             MRPublisherResponse res = pub.sendBatchWithResponse(); // As per dmaap-client code understanding, this need not be called but for some reason the messages are not pushed unless this is called
 112             LOG.debug("Response message = {} ", res.toString());
 113         }
 114     }
 115
 116     private void pauseThread() throws InterruptedException {
 117         if (fetchPause > 0) {
 118             LOG.debug("No data yet to publish.  Pausing {} ms before retry ", fetchPause);
 119             Thread.sleep(fetchPause);
 120         } else {
 121             LOG.debug("No data yet to publish. No fetch pause specified - retrying immediately");
 122         }
 123     }
 124
 125     public static void stopPublisher() throws IOException, InterruptedException {
 126         closePublisher = true;
 127         pub.close(100, TimeUnit.MILLISECONDS); // Send any remaining messages and close
 128     }
 129 }