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
  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, generalCfg.getMessageSentThreadOccurrence());
  67
  68                 createPublisher(publisherProperties);
  69         }
  70
  71         public MRBatchingPublisher createPublisher(Properties publisherProperties) {
  72
  73                 try {
  74                         pub = MRClientFactory.createBatchingPublisher(publisherProperties, false);
  75                         return pub;
  76                 } catch (IOException e) {
  77                         LOG.info("Exception while creating a publisher", e);
  78
  79                 }
  80                 return null;
  81         }
  82
  83         public void publishMessage(MRBatchingPublisher pub, String msg) {
  84                 LOG.info("Publishing message {} - ", msg);
  85                 try {
  86                         pub.send(msg);
  87                 } catch (IOException e) {
  88                         LOG.info("Exception while publishing a mesage ", e);
  89                 }
  90         }
  91
  92         public MRBatchingPublisher getPublisher() {
  93                 return pub;
  94         }
  95
  96         public void run() {
  97
  98                 while (!closePublisher) {
  99                         try {
 100                                 if (stateObjects.size() > 0) {
 101                                         JSONObject obj = ((LinkedList<JSONObject>) stateObjects).removeFirst();
 102                                         publishMessage(getPublisher(), obj.toString());
 103                                 } else {
 104                                         pauseThread();
 105                                 }
 106                         } catch(Exception ex) {
 107                                 LOG.error("Exception while publishing message, ignoring and continuing ... ", ex);
 108                         }
 109
 110                         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
 111                         LOG.debug("Response message = {} ",res.toString());
 112                 }
 113         }
 114
 115         private void pauseThread() throws InterruptedException {
 116                 if (fetchPause > 0) {
 117                         LOG.debug("No data yet to publish.  Pausing {} ms before retry ", fetchPause);
 118                         Thread.sleep(fetchPause);
 119                 } else {
 120                         LOG.debug("No data yet to publish. No fetch pause specified - retrying immediately");
 121                 }
 122         }
 123
 124         public static void stopPublisher() throws IOException, InterruptedException {
 125                 closePublisher = true;
 126                 pub.close(100, TimeUnit.MILLISECONDS); // Send any remaining messages and close
 127         }
 128 }