gerrit.onap Code Review - ccsdk/features.git/blob

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